diff --git a/.lane/attic/packages/channel/src/channel.ts/01M4XQJSWBSNDX2QZ1CQRBB9FH-pi-settles-a-run-s-inputs-in.md b/.lane/attic/packages/channel/src/channel.ts/01M4XQJSWBSNDX2QZ1CQRBB9FH-pi-settles-a-run-s-inputs-in.md new file mode 100644 index 00000000..175f67a3 --- /dev/null +++ b/.lane/attic/packages/channel/src/channel.ts/01M4XQJSWBSNDX2QZ1CQRBB9FH-pi-settles-a-run-s-inputs-in.md @@ -0,0 +1,8 @@ +--- +id: 01M4XQJSWBSNDX2QZ1CQRBB9FH +anchor: 'fn #ended' +created: 2026-10-14T17:32:54Z +norm: '1' +--- + +pi settles a run's inputs in the same commit that removes LiveDoc.run; read the outcome from that commit's submission changes, not from the transcript diff --git a/.lane/attic/packages/channel/src/channel.ts/01M4XQK5G3879T8FK6RJH5RCFF-pi-settles-a-run-s-inputs-in.md b/.lane/attic/packages/channel/src/channel.ts/01M4XQK5G3879T8FK6RJH5RCFF-pi-settles-a-run-s-inputs-in.md new file mode 100644 index 00000000..4d50791f --- /dev/null +++ b/.lane/attic/packages/channel/src/channel.ts/01M4XQK5G3879T8FK6RJH5RCFF-pi-settles-a-run-s-inputs-in.md @@ -0,0 +1,8 @@ +--- +id: 01M4XQK5G3879T8FK6RJH5RCFF +anchor: '#ended' +created: 2026-10-14T17:33:06Z +norm: '1' +--- + +pi settles a run's inputs in the same commit that removes LiveDoc.run; read the outcome from that commit's submission changes, not from the transcript diff --git a/.lane/attic/packages/channel/src/channel.ts/01M4XQK9XWF9WF94C1AJP4MSKF-pi-settles-a-run-s-inputs-in.md b/.lane/attic/packages/channel/src/channel.ts/01M4XQK9XWF9WF94C1AJP4MSKF-pi-settles-a-run-s-inputs-in.md new file mode 100644 index 00000000..38ebd536 --- /dev/null +++ b/.lane/attic/packages/channel/src/channel.ts/01M4XQK9XWF9WF94C1AJP4MSKF-pi-settles-a-run-s-inputs-in.md @@ -0,0 +1,8 @@ +--- +id: 01M4XQK9XWF9WF94C1AJP4MSKF +anchor: 'async #ended' +created: 2026-10-14T17:33:10Z +norm: '1' +--- + +pi settles a run's inputs in the same commit that removes LiveDoc.run; read the outcome from that commit's submission changes, not from the transcript diff --git a/.lane/attic/packages/channel/src/channel.ts/01M4XQKEF88070PYNF6AYDVZPF-a-run-s-end-is-read-from-the.md b/.lane/attic/packages/channel/src/channel.ts/01M4XQKEF88070PYNF6AYDVZPF-a-run-s-end-is-read-from-the.md new file mode 100644 index 00000000..824e1c58 --- /dev/null +++ b/.lane/attic/packages/channel/src/channel.ts/01M4XQKEF88070PYNF6AYDVZPF-a-run-s-end-is-read-from-the.md @@ -0,0 +1,8 @@ +--- +id: 01M4XQKEF88070PYNF6AYDVZPF +anchor: subscribeCommits +created: 2026-10-14T17:33:15Z +norm: '1' +--- + +a run's end is read from the same commit that removes LiveDoc.run: pi settles its inputs there (done / aborted -> stopped / else failed); don't scan transcripts for it diff --git a/.lane/attic/packages/channel/src/channel.ts/01M4XQKK8PW5KE1QZN53ZM6DY8-a-run-s-end-is-read-from-the.md b/.lane/attic/packages/channel/src/channel.ts/01M4XQKK8PW5KE1QZN53ZM6DY8-a-run-s-end-is-read-from-the.md new file mode 100644 index 00000000..7e23a142 --- /dev/null +++ b/.lane/attic/packages/channel/src/channel.ts/01M4XQKK8PW5KE1QZN53ZM6DY8-a-run-s-end-is-read-from-the.md @@ -0,0 +1,8 @@ +--- +id: 01M4XQKK8PW5KE1QZN53ZM6DY8 +anchor: static async open +created: 2026-10-14T17:33:20Z +norm: '1' +--- + +a run's end is read from the same commit that removes LiveDoc.run: pi settles its inputs there (done / aborted -> stopped / else failed); don't scan transcripts for it diff --git a/.lane/attic/packages/channel/src/room.ts/01M4XQJSWJMMPMH6PYZ4TWVZYF-subagent-and-cross-channel-i.md b/.lane/attic/packages/channel/src/room.ts/01M4XQJSWJMMPMH6PYZ4TWVZYF-subagent-and-cross-channel-i.md new file mode 100644 index 00000000..cb018f56 --- /dev/null +++ b/.lane/attic/packages/channel/src/room.ts/01M4XQJSWJMMPMH6PYZ4TWVZYF-subagent-and-cross-channel-i.md @@ -0,0 +1,8 @@ +--- +id: 01M4XQJSWJMMPMH6PYZ4TWVZYF +anchor: fn isAgent +created: 2026-10-14T17:32:54Z +norm: '1' +--- + +subagent and cross-channel inputs are authored agent@...; runs they start must not notify a person diff --git a/.lane/memory/apps/app/src/attention.ts/01M4XQJSWS1YHDS9TRD27C41WG-told-persists-per-host-and-u.md b/.lane/memory/apps/app/src/attention.ts/01M4XQJSWS1YHDS9TRD27C41WG-told-persists-per-host-and-u.md new file mode 100644 index 00000000..6aba8698 --- /dev/null +++ b/.lane/memory/apps/app/src/attention.ts/01M4XQJSWS1YHDS9TRD27C41WG-told-persists-per-host-and-u.md @@ -0,0 +1,12 @@ +--- +id: 01M4XQJSWS1YHDS9TRD27C41WG +anchor: '@file' +created: 2026-10-14T17:32:54Z +norm: '1' +sig: 9d222e7bbdcbbb9b +body_hash: 2de2b2790d826036 +raw_hash: b5ee961f9825a9a0 +lines: 1-102 +--- + +`told` persists per host and user so reloads and reconnects never re-notify the same run end; `since` keeps history from lighting up on a first visit diff --git a/.lane/memory/packages/channel/src/channel.ts/01M4XQKND0900E94CG3BXBEJP7-a-run-s-end-is-read-from-the.md b/.lane/memory/packages/channel/src/channel.ts/01M4XQKND0900E94CG3BXBEJP7-a-run-s-end-is-read-from-the.md new file mode 100644 index 00000000..3bcab79a --- /dev/null +++ b/.lane/memory/packages/channel/src/channel.ts/01M4XQKND0900E94CG3BXBEJP7-a-run-s-end-is-read-from-the.md @@ -0,0 +1,12 @@ +--- +id: 01M4XQKND0900E94CG3BXBEJP7 +anchor: class Channel +created: 2026-10-14T17:33:22Z +norm: '1' +sig: c905677fa4be33c1 +body_hash: c04286fc994309b7 +raw_hash: 0c9639c44cd13600 +lines: 114-792 +--- + +a run's end is read from the same commit that removes LiveDoc.run: pi settles its inputs there (done / aborted -> stopped / else failed); don't scan transcripts for it diff --git a/.lane/memory/packages/channel/src/room.ts/01M4XQK5FJQNG3VN0KFFCCYX3R-subagent-and-cross-channel-i.md b/.lane/memory/packages/channel/src/room.ts/01M4XQK5FJQNG3VN0KFFCCYX3R-subagent-and-cross-channel-i.md new file mode 100644 index 00000000..8b246cab --- /dev/null +++ b/.lane/memory/packages/channel/src/room.ts/01M4XQK5FJQNG3VN0KFFCCYX3R-subagent-and-cross-channel-i.md @@ -0,0 +1,12 @@ +--- +id: 01M4XQK5FJQNG3VN0KFFCCYX3R +anchor: function isAgent +created: 2026-10-14T17:33:06Z +norm: '1' +sig: b5c0961e8f085c4e +body_hash: 3956f4ef2be16694 +raw_hash: 1240d13ae4e3898c +lines: 23-25 +--- + +subagent and cross-channel inputs are authored agent@...; runs they start must not notify a person diff --git a/apps/app/src/app.tsx b/apps/app/src/app.tsx index a8aca23d..25d81f6c 100644 --- a/apps/app/src/app.tsx +++ b/apps/app/src/app.tsx @@ -24,6 +24,7 @@ import { IconHash, IconPlus } from "@ace/ui/icons"; import type { Hello, HostRequest, Listing, MoveTarget, People, Project } from "@ace/host/protocol"; import { chosen, deployed, remember } from "./address"; +import { glow, type Outcome, useAttention } from "./attention"; import { Channel, ChannelDetails, type ChannelDraft } from "./channel"; import { Dashboard } from "./dashboard"; import { Github, GithubSidebar } from "./github"; @@ -107,7 +108,7 @@ const INACTIVE_AFTER = 6 * 60 * 60_000; const ALL_PROJECTS = "all"; const RECENT_CHANNELS = 10; -function row(channel: Listing, user: string, host: string): SidebarRow { +function row(channel: Listing, user: string, host: string, outcome?: Outcome): SidebarRow { return { uid: channel.id, kind: "session", @@ -130,7 +131,8 @@ function row(channel: Listing, user: string, host: string): SidebarRow { ? "offline" : "idle", agent: channel.state === "running" && channel.busy ? "thinking" : "idle", - unreadCount: 0, + unreadCount: outcome === "done" ? 1 : 0, + needsAttention: outcome === "failed", mentionCount: 0, lastActivityAt: Math.floor((channel.active || channel.created) / 1000), online: [], @@ -153,13 +155,14 @@ function projectGroups( people: People, now: number, collapsed: Partial>, + attention: Map, ): SessionSidebarGroup[] { const mine: SidebarRow[] = []; const inactive: SidebarRow[] = []; const archived: SidebarRow[] = []; const team = new Map(); for (const channel of channels) { - const next = row(channel, hello.user, hello.host); + const next = row(channel, hello.user, hello.host, attention.get(channel.id)); const stored = channel.state === "archived"; if (channel.owner === hello.user) { const idle = !channel.busy && now - (channel.active || channel.created) > INACTIVE_AFTER; @@ -297,7 +300,12 @@ export function App() { org: value.repo?.split("/")[0] || "", })), ]; - let groups = projectGroups(visible, hello, people, now, collapsed); + const { attention, pulse } = useAttention( + channels, + hello.user, + page === "channels" ? channel?.id : undefined, + ); + let groups = projectGroups(visible, hello, people, now, collapsed, attention); if (allProjects) { const byProject = new Map(); @@ -313,7 +321,9 @@ export function App() { const rows = (byProject.get(value.id) || []).sort((a, b) => (b.active || b.created) - (a.active || a.created) || b.created - a.created || b.id.localeCompare(a.id) - ).slice(0, RECENT_CHANNELS).map((value) => row(value, hello.user, hello.host)); + ).slice(0, RECENT_CHANNELS).map((value) => + row(value, hello.user, hello.host, attention.get(value.id)) + ); return { id, label: label(value, hello.host), @@ -484,7 +494,8 @@ export function App() { onLeftChange={setLeft} leftWidth={width} onLeftWidthChange={setWidth} - loading={status === "connecting"} + loading={status === "connecting" || pulse === "on"} + {...(pulse && glow.props)} onDoubleClick={titlebar} nav={ ; told: Record }; + +export type Outcome = Exclude; + +/** The pulse holds its glow this long, then keeps these props until it has faded out. */ +export const glow = { hold: 1800, props: { glowOpacity: 0.6, glowFadeIn: 0.8, glowFadeOut: 1.3 } }; + +const EVENTS = ["focus", "blur", "visibilitychange"]; + +function subscribe(listener: () => void) { + for (const event of EVENTS) addEventListener(event, listener); + return () => { + for (const event of EVENTS) removeEventListener(event, listener); + }; +} + +const isWatching = () => document.visibilityState === "visible" && document.hasFocus(); + +/** A stopped run was a participant's own doing, so it needs no attention. */ +function pending(channel: Listing, user: string, state: State): Outcome | undefined { + const { ran } = channel; + if (!ran || ran.author !== user || ran.outcome === "stopped") return; + if (ran.at <= Math.max(state.since, state.seen[channel.id] || 0)) return; + return ran.outcome; +} + +/** + * Runs this person invoked that ended while they were not watching the channel. Each end pulses + * the window's glow or sends one desktop notification, then marks its row until they open it. + */ +export function useAttention(channels: Listing[], user: string, open: string | undefined) { + const [state, setState] = useLocalStorage( + `ace:attention:${host.url}:${user}`, + null, + ); + const watching = useSyncExternalStore(subscribe, isWatching); + const [pulse, setPulse] = useState<"on" | "fading">(); + + useEffect(() => { + if (user && !state) setState({ since: Date.now(), seen: {}, told: {} }); + }, [user, state, setState]); + + useEffect(() => { + if (!user || !state) return; + const seen: Record = {}; + const told: Record = {}; + for (const channel of channels) { + const { id, ran } = channel; + if (!ran) continue; + if (watching && id === open) { + if ((state.seen[id] || 0) < ran.at) seen[id] = ran.at; + continue; + } + const outcome = pending(channel, user, state); + if (!outcome || ran.at <= (state.told[id] || 0)) continue; + told[id] = ran.at; + if (watching) setPulse("on"); + else { + desktop?.notify(channel.name, outcome === "failed" ? "Agent run failed" : "Agent finished") + .catch(() => {}); + } + } + if (!Object.keys(seen).length && !Object.keys(told).length) return; + setState((value) => + value && { ...value, seen: { ...value.seen, ...seen }, told: { ...value.told, ...told } } + ); + }, [channels, user, state, setState, watching, open]); + + useEffect(() => { + if (!pulse) return; + const timer = setTimeout( + () => setPulse(pulse === "on" ? "fading" : undefined), + pulse === "on" ? glow.hold : glow.props.glowFadeOut * 1000, + ); + return () => clearTimeout(timer); + }, [pulse]); + + const attention = useMemo(() => { + const found = new Map(); + if (!state) return found; + for (const channel of channels) { + const outcome = pending(channel, user, state); + if (outcome) found.set(channel.id, outcome); + } + return found; + }, [channels, user, state]); + + return { attention, pulse }; +} diff --git a/apps/app/src/desktop.ts b/apps/app/src/desktop.ts index 20358db6..2c68b9ca 100644 --- a/apps/app/src/desktop.ts +++ b/apps/app/src/desktop.ts @@ -22,6 +22,7 @@ export const desktop = view && token helper: (action: HelperAction) => rpc.request.helper({ token, action }), native: (action: NativeAction) => rpc.request.native({ token, action }), updates: (action: UpdateAction) => rpc.request.updates({ token, action }), + notify: (title: string, body: string) => rpc.request.notify({ token, title, body }), } : undefined; diff --git a/apps/desktop/src/index.ts b/apps/desktop/src/index.ts index 08d2e04d..24513ad2 100644 --- a/apps/desktop/src/index.ts +++ b/apps/desktop/src/index.ts @@ -64,6 +64,11 @@ const rpc = BrowserView.defineRPC({ authorize(token); return native.act(action); }, + notify: ({ token, title, body }) => { + authorize(token); + Utils.showNotification({ title, body }); + return null; + }, }, messages: {}, }, diff --git a/apps/desktop/src/protocol.ts b/apps/desktop/src/protocol.ts index d6257c3c..a22e165e 100644 --- a/apps/desktop/src/protocol.ts +++ b/apps/desktop/src/protocol.ts @@ -66,6 +66,7 @@ export type DesktopRPC = { helper: { params: { token: string; action: HelperAction }; response: HelperState }; native: { params: { token: string; action: NativeAction }; response: NativeState }; updates: { params: { token: string; action: UpdateAction }; response: UpdateState }; + notify: { params: { token: string; title: string; body: string }; response: null }; }; messages: {}; }; diff --git a/apps/host/src/catalog.ts b/apps/host/src/catalog.ts index 95290ef5..b335c5f5 100644 --- a/apps/host/src/catalog.ts +++ b/apps/host/src/catalog.ts @@ -12,7 +12,7 @@ import { import { userInfo } from "node:os"; import { join } from "node:path"; -import { type Metadata, type ModelRef, validateName } from "@ace/channel"; +import { type Metadata, type ModelRef, type RunEnd, validateName } from "@ace/channel"; import { config } from "./config"; import { login } from "./tailnet"; @@ -36,6 +36,8 @@ export type Listing = { active?: number; /** Rebuildable live-run projection, not history; valid only while its source is connected. */ busy?: boolean; + /** The latest invoked run's end, a hint for its author's attention; pi keeps the run itself. */ + ran?: RunEnd; archived?: boolean; /** Base URL of the service hosting the channel; this host serves its workspace. */ hosted?: string; @@ -185,6 +187,10 @@ export function busy(id: string, busy: boolean): void { write({ ...record, busy }); } +export function ran(id: string, ran: RunEnd): void { + write({ ...read(id), ran }); +} + export function list(): Listing[] { if (!existsSync(root)) return []; return readdirSync(root) diff --git a/apps/host/src/gateway.ts b/apps/host/src/gateway.ts index 124b8efb..b19a4b28 100644 --- a/apps/host/src/gateway.ts +++ b/apps/host/src/gateway.ts @@ -68,8 +68,19 @@ const deleting = new Set(); function local(): Listing[] { return catalog.list().map((record) => { const state = record.archived ? "archived" : isRunning(record.id) ? "running" : "dormant"; - const { id, name: channel, summary, revision, owner, project, model, created, active, hosted } = - record; + const { + id, + name: channel, + summary, + revision, + owner, + project, + model, + created, + active, + ran, + hosted, + } = record; const listing: Listing = { id, host: name, @@ -84,6 +95,7 @@ function local(): Listing[] { busy: state === "running" && !!record.busy && (!hosted || workspace.isConnected(id)), }; if (active) listing.active = active; + if (ran) listing.ran = ran; const { root, repo } = projects.checkout(project); listing.root = root; if (repo) listing.repo = repo; diff --git a/apps/host/src/protocol.ts b/apps/host/src/protocol.ts index 2dd2f373..3cbfbeb9 100644 --- a/apps/host/src/protocol.ts +++ b/apps/host/src/protocol.ts @@ -6,6 +6,7 @@ import type { GithubKind, ModelRef, Request, + RunEnd, } from "@ace/channel/protocol"; export const HOST_PROTOCOL = 9; @@ -59,6 +60,8 @@ export type Listing = { active?: number; /** Live channel-wide run projection; absent on older hosts, never authoritative. */ busy?: boolean; + /** The latest invoked run's end; absent on older hosts and before a channel's first one. */ + ran?: RunEnd; state: ChannelState; /** The hosting service's base URL, for a hosted channel; `host` is then its workspace. */ hosted?: string; diff --git a/apps/host/src/worker.ts b/apps/host/src/worker.ts index 01df2e08..68e44f24 100644 --- a/apps/host/src/worker.ts +++ b/apps/host/src/worker.ts @@ -55,6 +55,7 @@ const channel = await Channel.open({ onMetadata: (value) => catalog.metadata(id, value), onActivity: (at) => catalog.activity(id, at), onBusy: (busy) => catalog.busy(id, busy), + onRun: (ran) => catalog.ran(id, ran), owner: record.owner, project: record.project, model: record.model, diff --git a/apps/host/src/workspace.ts b/apps/host/src/workspace.ts index f9474d56..5ea05353 100644 --- a/apps/host/src/workspace.ts +++ b/apps/host/src/workspace.ts @@ -45,6 +45,7 @@ function link(record: catalog.Listing): () => Promise { if ("metadata" in call) return catalog.metadata(record.id, call.metadata); if ("active" in call) return catalog.activity(record.id, call.active); if ("busy" in call) return catalog.busy(record.id, call.busy); + if ("ran" in call) return catalog.ran(record.id, call.ran); log("debug", "workspace.call", { channel: record.id, ...("cancel" in call diff --git a/docs/architecture.md b/docs/architecture.md index 794719bf..7879dd3e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -120,6 +120,11 @@ is one pi conversation in that Session. committed live-document changes maintain the aggregate and `onBusy` publishes only transitions. Streaming transcript watches are not needed for sidebar rows. Pi remains authoritative: the optional catalog and listing `busy` field is rebuildable, not a second run store or history format. +- **Run ends are a listing hint for attention.** When a chat's run ends, the channel reads the + settlement pi commits with it (answered, stopped, or failed) and the newest human input the run + answered. Runs only agents invoked, such as subagents' chats, publish nothing. `onRun` projects + the latest end into the catalog record and listing as `ran`, carried over the workspace socket + for hosted channels. Losing it loses only a pending hint; pi keeps the runs. `packages/channel` must stay runtime-neutral: no `node:*`, `bun:*`, or Workers imports. The host injects storage, models, and the execution environment. This keeps hosted channels (a channel @@ -444,7 +449,12 @@ channel stays listed while its host is asleep. The app shows a local channel who unreachable as offline, and reaches a hosted channel directly. Hosts find the directory through `ace directory ` and authenticate with the team's `ACE_SECRET`. -Lobbies (presence, notifications) are not built. +Lobbies (presence, notifications) are not built. Until then each app notices its own person's +finished and failed runs from listings: it keeps the run ends it has shown or announced per host +and participant in browser storage, starting from its first visit. A run end that arrives while +another channel is open pulses the window's glow; one that arrives while the window is unfocused +sends a desktop notification. The channel's row keeps a dot (finished) or attention mark (failed) +until the person opens it in a focused window. Stopped runs need no attention. ## The team diff --git a/packages/channel/src/channel.ts b/packages/channel/src/channel.ts index a5cc8dab..3d0d1afa 100644 --- a/packages/channel/src/channel.ts +++ b/packages/channel/src/channel.ts @@ -3,6 +3,7 @@ import { BACKGROUND_CONTEXT as context } from "@earendil-works/chord/context"; import { AgentDoc, type AgentEvent, + type CommitChange, type Conversation, type ConversationId, type ConversationRecord, @@ -14,9 +15,11 @@ import { Harness, InboxDoc, LiveDoc, + type LiveState, ROOT_CONVERSATION_ID, type Storage, type SubmissionId, + type SubmissionRecord, ToolResultEntry, UsageDoc, type UsageState, @@ -54,6 +57,7 @@ import type { Metadata, ModelRef, Request, + RunEnd, Usage, } from "./protocol"; import { queue } from "./queue"; @@ -100,6 +104,8 @@ export type Options = { onActivity?(at: number): void; /** Rebuildable channel-wide projection of pi live runs, including child chats. */ onBusy?(busy: boolean): void; + /** Publish each run a person invoked as it ends, for their attention. */ + onRun?(run: RunEnd): void; log: Log; }; @@ -113,7 +119,8 @@ export class Channel { /** Committed settings, maintained from pi's commit stream before native dispatch. */ #settings: Settings; #active = 0; - #runs = new Set(); + /** Each live run's inputs, from pi's `LiveDoc`. */ + #runs = new Map(); #publishedBusy?: boolean; #unsubscribe?: () => void; #admissions = new Map>(); @@ -198,6 +205,7 @@ export class Channel { // and subscription cannot miss a run transition. LiveDoc forks start empty in pi. await channel.#snapshotBusy(); channel.#unsubscribe = harness.subscribeCommits(({ changes }) => { + const ended: [ConversationId, readonly SubmissionId[]][] = []; for (const change of changes) { if (change.type === "entry" && change.value.model?.[0]?.role === "assistant") { channel.#caption(change.value.conversationId); @@ -212,11 +220,17 @@ export class Channel { if (change.record.kind !== LiveDoc.definition.kind) continue; const { scope } = change.record; if (scope.kind !== "conversation") continue; - if (change.value?.run) channel.#runs.add(scope.conversationId); - else if (channel.#runs.delete(scope.conversationId)) { + const { run } = (change.value || {}) as LiveState; + const inputs = channel.#runs.get(scope.conversationId); + if (run) channel.#runs.set(scope.conversationId, run.inputs); + else if (inputs) { + channel.#runs.delete(scope.conversationId); channel.#caption(scope.conversationId); + ended.push([scope.conversationId, inputs]); } } + // Ending a run settles its inputs in the same commit, in no particular change order. + for (const [chat, inputs] of ended) void channel.#ended(chat, inputs, changes); // Publish once per atomic commit, not once per chat or streaming delta. channel.#publishBusy(); if (changes.some((change) => change.type === "entry")) channel.#activity(); @@ -366,7 +380,7 @@ export class Channel { ); for (const { id } of page.items) { const live = await this.#harness.snapshot(LiveDoc, id, context); - if (live?.run) this.#runs.add(id); + if (live?.run) this.#runs.set(id, live.run.inputs); } cursor = page.next; } while (cursor); @@ -384,6 +398,37 @@ export class Channel { } } + async #ended( + chat: ConversationId, + inputs: readonly SubmissionId[], + changes: readonly CommitChange[], + ): Promise { + const settled = new Map(); + for (const change of changes) { + if (change.type === "submission") settled.set(change.value.id, change.value); + } + const last = settled.get(inputs.at(-1)!); + if (!last) return; + const outcome = last.status === "done" + ? "done" + : last.reason === "aborted" + ? "stopped" + : "failed"; + try { + // The newest human input names who is waiting; agents' inputs are their parents' work. + const author = await this.#harness.commit(async (tx) => { + for (const id of inputs.toReversed()) { + const entry = settled.get(id)?.entry; + const human = entry === undefined ? undefined : room.human((await tx.entry(entry))!); + if (human && !room.isAgent(human.author)) return human.author; + } + }, context); + if (author) this.#options.onRun?.({ chat, at: Date.now(), outcome, author }); + } catch (error) { + this.#log("warn", "run.publish", failure(error)); + } + } + /** Durably abort every chat's work, including background subagents, so reopening resumes none of it. */ async kill(): Promise { const { chats } = await this.info(); diff --git a/packages/channel/src/protocol.ts b/packages/channel/src/protocol.ts index 98874134..f8853db0 100644 --- a/packages/channel/src/protocol.ts +++ b/packages/channel/src/protocol.ts @@ -24,6 +24,17 @@ export type EntryId = number; export type Metadata = { name: string; summary: string; revision: number }; +/** + * The end of a run a person invoked. `stopped` means a participant aborted it; `author` is the + * participant whose input the run answered last. + */ +export type RunEnd = { + chat: ChatId; + at: number; + outcome: "done" | "failed" | "stopped"; + author: string; +}; + export type ChannelInfo = Metadata & { id: string; /** Directory of the project checkout lanes are created from. */ diff --git a/packages/channel/src/room.ts b/packages/channel/src/room.ts index 74e6a3f5..bb88df89 100644 --- a/packages/channel/src/room.ts +++ b/packages/channel/src/room.ts @@ -19,6 +19,11 @@ export function isAuthor(value: string): boolean { return AUTHOR.test(value); } +/** Subagent tasks and cross-channel messages carry `agent@` authors. */ +export function isAgent(author: string): boolean { + return author.startsWith("agent@"); +} + /** * Agents see every participant's messages as `author: text`. Invocations are pi inputs, which carry * only content, so the author prefix is also how an input's author is recovered. diff --git a/packages/channel/src/workspace.ts b/packages/channel/src/workspace.ts index 40279d9f..c45c69ff 100644 --- a/packages/channel/src/workspace.ts +++ b/packages/channel/src/workspace.ts @@ -16,7 +16,7 @@ import { import { type Browser, type BrowserRequest, type BrowserResult, isBrowserAction } from "./browser"; import { type Desktop, type DesktopRequest, type DesktopResult, isDesktopAction } from "./desktop"; -import type { Metadata } from "./protocol"; +import type { Metadata, RunEnd } from "./protocol"; declare function btoa(data: string): string; declare function atob(data: string): string; @@ -30,9 +30,12 @@ export type Call = | { cancel: CallId }; /** Metadata and activity are projected by the workspace host even when no client watches the channel. */ -export type WorkspaceMessage = Call | { metadata: Metadata } | { active: number } | { - busy: boolean; -}; +export type WorkspaceMessage = + | Call + | { metadata: Metadata } + | { active: number } + | { busy: boolean } + | { ran: RunEnd }; /** Workspace to channel. */ export type Reply = diff --git a/services/channel/src/object.ts b/services/channel/src/object.ts index 4469f57b..6715f859 100644 --- a/services/channel/src/object.ts +++ b/services/channel/src/object.ts @@ -132,6 +132,7 @@ export class HostedChannel extends DurableObject { ...config, onMetadata: (metadata) => this.#project({ metadata }), onActivity: (active) => this.#project({ active }), + onRun: (ran) => this.#project({ ran }), onBusy: (busy) => { this.#reside(busy); this.#project({ busy });