From ca61d7827eceeb5b0608feb02c8ad7418206383e Mon Sep 17 00:00:00 2001 From: qqqys Date: Thu, 2 Jul 2026 17:04:07 +0800 Subject: [PATCH] feat(channels): add identity and task lifecycle metadata (#6105) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * chore: ignore local worktrees * docs(channels): design identity and task lifecycle p0 * docs(channels): plan identity and task lifecycle p0 * feat(channels): add identity and task lifecycle metadata * fix(channels): suppress cancelled tool call lifecycle * fix(channels): cover loop lifecycle metadata * fix(channels): harden lifecycle event edges * fix(channels): finalize cancelled lifecycle before cleanup * fix(channels): address lifecycle review suggestions * fix(channels): close lifecycle cancellation races * fix(channels): separate pending cancel state * fix(channels): order clear cancellation lifecycle * fix(channels): suppress loop chunks during pending cancel * test(channels): use active session in cancel regression * fix(channels): sanitize lifecycle tool fields * fix(channels): route shared tool call lifecycle * fix(channels): preserve pending cancel intent * fix(channels): preserve responses after failed cancel * fix(channels): tighten lifecycle cancellation reasons * fix(channels): close lifecycle cancel races and validate identity config Address the outstanding review findings on #6105: - carry a typed reason on ChannelLoopSkippedError and report disabled loops as 'dropped' instead of 'timeout' - treat a turn as committed once delivery starts: /cancel re-checks deliveryStarted after the cancel RPC settles, and neither prompt path lets a late-settling cancel rewrite a delivered turn into cancelled (or follow a /clear cancellation with completed) - tag failed lifecycle events with phase: agent vs delivery - validate identity/memoryScope shape at config parse time instead of throwing an opaque TypeError on the first prompt of every session - guard onPromptStart hooks, pass job.id to loop onPromptStart/End, append the boundary block after operator instructions, gate /who + /status identity lines on configured identity/memoryScope, cache the boundary prompt, and route error logs through lifecycleError() Co-authored-by: Qwen-Coder * fix(channels): keep failed terminal event when cancel settles post-delivery Self-review follow-up: once delivery started, the catch paths no longer reconcile a pending cancel — a late-resolving cancel RPC used to flip cancelled=true there, suppressing the failed emit while the /cancel handler (seeing deliveryStarted) also declined to emit, leaving a started task with no terminal lifecycle event at all. Co-authored-by: Qwen-Coder * fix(channels): hold streamed chunks while a cancel is pending Chunks arriving while a /cancel RPC is in flight were pushed straight into the BlockStreamer, which can send a block on a size/paragraph threshold before the cancel resolves — leaking output a successful cancel can never recall. Hold the pending-window chunks instead: replay them (block streaming + text_chunk transcript) when the cancel fails, discard them when it succeeds. onResponseChunk stays live through the window so adapter-accumulated display state has no permanent hole on a failed cancel; adapters gate visible updates on their own stop flags. Also stop passing the loop job id to onPromptStart/onPromptEnd (and the /clear eviction path): the hook contract is inbound platform message ids, and adapters act on them — cards and reactions keyed to a fake id. Lifecycle events still carry job.id for correlation. Co-authored-by: Qwen-Coder * fix(channels): defer adapter chunks while cancel is pending --------- Co-authored-by: Shaojin Wen Co-authored-by: Qwen-Coder --- .gitignore | 1 + ...7-01-channel-p0-identity-task-lifecycle.md | 735 +++++++ ...annel-p0-identity-task-lifecycle-design.md | 272 +++ .../channels/base/src/ChannelBase.test.ts | 1849 ++++++++++++++++- packages/channels/base/src/ChannelBase.ts | 497 ++++- .../channels/base/src/ChannelLoopScheduler.ts | 12 +- packages/channels/base/src/index.ts | 9 + packages/channels/base/src/types.ts | 75 + .../src/commands/channel/config-utils.test.ts | 49 + .../cli/src/commands/channel/config-utils.ts | 61 + packages/cli/src/commands/channel/runtime.ts | 2 +- .../cli/src/commands/channel/start.test.ts | 5 +- 12 files changed, 3524 insertions(+), 43 deletions(-) create mode 100644 docs/superpowers/plans/2026-07-01-channel-p0-identity-task-lifecycle.md create mode 100644 docs/superpowers/specs/2026-07-01-channel-p0-identity-task-lifecycle-design.md diff --git a/.gitignore b/.gitignore index 9ac29593eb..91edb3de5d 100644 --- a/.gitignore +++ b/.gitignore @@ -120,6 +120,7 @@ coverage/ # Dev symlink: qc-helper bundled skill docs (created by scripts/dev.js) packages/core/src/skills/bundled/qc-helper/docs tmp/ +.worktrees/ # code graph skills .venv diff --git a/docs/superpowers/plans/2026-07-01-channel-p0-identity-task-lifecycle.md b/docs/superpowers/plans/2026-07-01-channel-p0-identity-task-lifecycle.md new file mode 100644 index 0000000000..225ff470fe --- /dev/null +++ b/docs/superpowers/plans/2026-07-01-channel-p0-identity-task-lifecycle.md @@ -0,0 +1,735 @@ +# Channel P0 Identity And Task Lifecycle Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add channel identity/memory-scope metadata and a base task lifecycle hook for channel-resident agents. + +**Architecture:** Keep the behavior inside `@qwen-code/channel-base`. Add typed metadata to `types.ts`, derive defaults in `ChannelBase`, inject a first-session boundary note alongside existing instructions, and emit lifecycle events through a protected no-op hook that adapters can override. + +**Tech Stack:** TypeScript, Vitest, existing `ChannelBase` / `ChannelConfig` / `ChannelAgentBridge` infrastructure. + +--- + +## File Structure + +- Modify `packages/channels/base/src/types.ts`: add identity, memory-scope, runtime metadata, and lifecycle event types. +- Modify `packages/channels/base/src/ChannelBase.ts`: derive metadata, inject prompt boundary, expose status metadata, emit lifecycle events. +- Modify `packages/channels/base/src/ChannelBase.test.ts`: add focused tests using the existing `TestChannel` fixture. +- No changes to core memory files, daemon events, or platform adapter UI. + +## Task 1: Add Channel Metadata And Lifecycle Types + +**Files:** +- Modify: `packages/channels/base/src/types.ts` +- Test: `packages/channels/base/src/ChannelBase.test.ts` + +- [ ] **Step 1: Write failing type-driven tests in `ChannelBase.test.ts`** + +Add `taskEvents` to `TestChannel` and two tests near the existing instruction tests: + +```ts +import type { + ChannelConfig, + ChannelTaskLifecycleEvent, + Envelope, +} from './types.js'; + +class TestChannel extends ChannelBase { + taskEvents: ChannelTaskLifecycleEvent[] = []; + + protected override onTaskLifecycle(event: ChannelTaskLifecycleEvent): void { + this.taskEvents.push(event); + } +} + +it('derives default channel identity and memory metadata for task lifecycle events', async () => { + const ch = createChannel(); + + await ch.handleInbound(envelope({ messageId: 'm-1' })); + + expect(ch.taskEvents[0]).toMatchObject({ + type: 'started', + channelName: 'test-chan', + chatId: 'chat1', + sessionId: 's-1', + messageId: 'm-1', + identity: { + id: 'channel:test-chan', + displayName: 'test-chan', + }, + memoryScope: { + namespace: 'channel:test-chan', + mode: 'metadata-only', + }, + }); +}); + +it('uses configured channel identity and memory namespace in lifecycle metadata', async () => { + const ch = createChannel({ + identity: { + id: 'ops-agent', + displayName: 'Ops Agent', + description: 'Coordinates repository operations.', + }, + memoryScope: { + namespace: 'qwen-tag:ops', + mode: 'metadata-only', + }, + }); + + await ch.handleInbound(envelope()); + + expect(ch.taskEvents[0]).toMatchObject({ + identity: { + id: 'ops-agent', + displayName: 'Ops Agent', + description: 'Coordinates repository operations.', + }, + memoryScope: { + namespace: 'qwen-tag:ops', + mode: 'metadata-only', + }, + }); +}); +``` + +- [ ] **Step 2: Run tests to verify type failures** + +Run: + +```bash +cd packages/channels/base +npx vitest run src/ChannelBase.test.ts +``` + +Expected: fails because `ChannelTaskLifecycleEvent`, `identity`, `memoryScope`, and `onTaskLifecycle` do not exist. + +- [ ] **Step 3: Add types in `types.ts`** + +Insert after `DispatchMode`: + +```ts +export interface ChannelIdentityConfig { + id?: string; + displayName?: string; + description?: string; +} + +export interface ChannelRuntimeIdentity { + id: string; + displayName: string; + description?: string; +} + +export type ChannelMemoryScopeMode = 'metadata-only'; + +export interface ChannelMemoryScopeConfig { + namespace?: string; + mode?: ChannelMemoryScopeMode; +} + +export interface ChannelRuntimeMemoryScope { + namespace: string; + mode: ChannelMemoryScopeMode; +} +``` + +Add to `ChannelConfig` after `instructions?: string;`: + +```ts + identity?: ChannelIdentityConfig; + memoryScope?: ChannelMemoryScopeConfig; +``` + +Import `ToolCallEvent` at the top: + +```ts +import type { ToolCallEvent } from './ChannelAgentBridge.js'; +``` + +Add before `ChannelPlugin`: + +```ts +interface ChannelTaskLifecycleBase { + channelName: string; + chatId: string; + sessionId: string; + messageId?: string; + identity: ChannelRuntimeIdentity; + memoryScope: ChannelRuntimeMemoryScope; +} + +export type ChannelTaskLifecycleEvent = + | (ChannelTaskLifecycleBase & { type: 'started' }) + | (ChannelTaskLifecycleBase & { type: 'text_chunk'; chunk: string }) + | (Omit & { + type: 'tool_call'; + toolCall: ToolCallEvent; + }) + | (ChannelTaskLifecycleBase & { + type: 'cancelled'; + reason: 'cancel_command' | 'clear' | 'steer'; + }) + | (ChannelTaskLifecycleBase & { type: 'completed' }) + | (ChannelTaskLifecycleBase & { type: 'failed'; error: string }); +``` + +- [ ] **Step 4: Add minimal runtime support in `ChannelBase.ts`** + +Update imports: + +```ts +import type { + ChannelConfig, + ChannelRuntimeIdentity, + ChannelRuntimeMemoryScope, + ChannelTaskLifecycleEvent, + DispatchMode, + Envelope, +} from './types.js'; +``` + +Add private fields after `protected name: string;`: + +```ts + private readonly identity: ChannelRuntimeIdentity; + private readonly memoryScope: ChannelRuntimeMemoryScope; +``` + +Set them in the constructor after `this.proxy = options?.proxy;`: + +```ts + this.identity = this.resolveIdentity(name, config); + this.memoryScope = this.resolveMemoryScope(name, config); +``` + +Add methods near other protected hooks: + +```ts + protected onTaskLifecycle(_event: ChannelTaskLifecycleEvent): void {} + + private emitTaskLifecycle(event: ChannelTaskLifecycleEvent): void { + try { + this.onTaskLifecycle(event); + } catch (err) { + process.stderr.write( + `[${this.name}] onTaskLifecycle threw for ${event.type} session ${event.sessionId}: ${ + err instanceof Error ? err.message : err + }\n`, + ); + } + } + + private resolveIdentity( + name: string, + config: ChannelConfig, + ): ChannelRuntimeIdentity { + return { + id: config.identity?.id || `channel:${name}`, + displayName: config.identity?.displayName || name, + ...(config.identity?.description + ? { description: config.identity.description } + : {}), + }; + } + + private resolveMemoryScope( + name: string, + config: ChannelConfig, + ): ChannelRuntimeMemoryScope { + return { + namespace: config.memoryScope?.namespace || `channel:${name}`, + mode: 'metadata-only', + }; + } +``` + +Emit `started` after `this.activePrompts.set(sessionId, promptState);`: + +```ts + this.emitTaskLifecycle({ + type: 'started', + channelName: this.name, + chatId: envelope.chatId, + sessionId, + messageId: envelope.messageId, + identity: this.identity, + memoryScope: this.memoryScope, + }); +``` + +- [ ] **Step 5: Run test to verify Task 1 passes** + +Run: + +```bash +cd packages/channels/base +npx vitest run src/ChannelBase.test.ts +``` + +Expected: both new metadata tests pass; existing tests still pass. + +## Task 2: Add Prompt Boundary And Status Visibility + +**Files:** +- Modify: `packages/channels/base/src/ChannelBase.ts` +- Test: `packages/channels/base/src/ChannelBase.test.ts` + +- [ ] **Step 1: Write failing prompt and status tests** + +Add tests near existing instruction and command tests: + +```ts +it('prepends channel boundary metadata before custom instructions once per session', async () => { + const ch = createChannel({ + instructions: 'Be concise.', + identity: { + id: 'ops-agent', + displayName: 'Ops Agent', + description: 'Coordinates repository operations.', + }, + memoryScope: { + namespace: 'qwen-tag:ops', + mode: 'metadata-only', + }, + }); + + await ch.handleInbound(envelope({ text: 'first' })); + await ch.handleInbound(envelope({ text: 'second' })); + + const prompt = vi.mocked(bridge.prompt).mock.calls[0]![1]; + expect(prompt).toContain('Channel identity:'); + expect(prompt).toContain('- id: ops-agent'); + expect(prompt).toContain('- display name: Ops Agent'); + expect(prompt).toContain( + '- description: Coordinates repository operations.', + ); + expect(prompt).toContain('Memory scope:'); + expect(prompt).toContain('- namespace: qwen-tag:ops'); + expect(prompt).toContain('- mode: metadata-only'); + expect(prompt).toContain('- storage isolation: not enforced by this version.'); + expect(prompt.indexOf('Channel identity:')).toBeLessThan( + prompt.indexOf('Be concise.'), + ); + + const secondPrompt = vi.mocked(bridge.prompt).mock.calls[1]![1]; + expect(secondPrompt).not.toContain('Channel identity:'); +}); + +it('/who and /status include channel identity and memory metadata', async () => { + const ch = createChannel({ + identity: { id: 'ops-agent', displayName: 'Ops Agent' }, + memoryScope: { namespace: 'qwen-tag:ops', mode: 'metadata-only' }, + }); + + await ch.handleInbound(envelope({ text: '/who' })); + await ch.handleInbound(envelope({ text: '/status' })); + + expect(ch.sent[0]!.text).toContain('Identity: Ops Agent'); + expect(ch.sent[0]!.text).toContain('Memory: qwen-tag:ops'); + expect(ch.sent[1]!.text).toContain('Identity: ops-agent'); + expect(ch.sent[1]!.text).toContain('Memory: metadata-only'); +}); +``` + +- [ ] **Step 2: Run tests to verify failures** + +Run: + +```bash +cd packages/channels/base +npx vitest run src/ChannelBase.test.ts +``` + +Expected: fails because prompt boundary and status lines are not implemented. + +- [ ] **Step 3: Implement boundary formatter in `ChannelBase.ts`** + +Add private method near metadata resolvers: + +```ts + private shouldPrependChannelBoundaryPrompt(): boolean { + return Boolean( + this.config.instructions || + this.config.identity || + this.config.memoryScope, + ); + } + + private channelBoundaryPrompt(): string { + const identityLines = [ + 'Channel identity:', + `- id: ${this.identity.id}`, + `- display name: ${this.identity.displayName}`, + ...(this.identity.description + ? [`- description: ${this.identity.description}`] + : []), + ]; + const memoryLines = [ + 'Memory scope:', + `- namespace: ${this.memoryScope.namespace}`, + `- mode: ${this.memoryScope.mode}`, + '- storage isolation: not enforced by this version.', + ]; + return [...identityLines, '', ...memoryLines].join('\n'); + } +``` + +Replace the existing instruction block: + +```ts + if (this.config.instructions && !this.instructedSessions.has(sessionId)) { + promptText = `${this.config.instructions}\n\n${promptText}`; + this.instructedSessions.add(sessionId); + } +``` + +with: + +```ts + if ( + this.shouldPrependChannelBoundaryPrompt() && + !this.instructedSessions.has(sessionId) + ) { + const prefix = this.config.instructions + ? `${this.channelBoundaryPrompt()}\n\n${this.config.instructions}` + : this.channelBoundaryPrompt(); + promptText = `${prefix}\n\n${promptText}`; + this.instructedSessions.add(sessionId); + } +``` + +- [ ] **Step 4: Add status lines** + +In `/who` lines, after `Channel: ${this.name}`, add: + +```ts + `Identity: ${this.identity.displayName}`, + `Memory: ${this.memoryScope.namespace}`, +``` + +In `/status` lines, after `Channel: ${this.name}`, add: + +```ts + `Identity: ${this.identity.id}`, + `Memory: ${this.memoryScope.mode}`, +``` + +- [ ] **Step 5: Run tests** + +Run: + +```bash +cd packages/channels/base +npx vitest run src/ChannelBase.test.ts +``` + +Expected: prompt boundary and status tests pass; existing instruction tests are updated if they expected custom instructions to be the only prefix. + +## Task 3: Emit Full Task Lifecycle Events + +**Files:** +- Modify: `packages/channels/base/src/ChannelBase.ts` +- Test: `packages/channels/base/src/ChannelBase.test.ts` + +- [ ] **Step 1: Write failing lifecycle tests** + +Add tests near existing streaming/cancel/dispatch tests: + +```ts +it('emits task lifecycle for chunks, tool calls, and completion', async () => { + vi.mocked(bridge.prompt).mockImplementation(async () => { + (bridge as unknown as EventEmitter).emit('textChunk', 's-1', 'hello '); + (bridge as unknown as EventEmitter).emit('toolCall', { + sessionId: 's-1', + toolCallId: 'tool-1', + kind: 'shell', + status: 'running', + }); + return 'agent response'; + }); + const ch = createChannel(); + + await ch.handleInbound(envelope({ messageId: 'm-1' })); + + expect(ch.taskEvents.map((event) => event.type)).toEqual([ + 'started', + 'text_chunk', + 'tool_call', + 'completed', + ]); + expect(ch.taskEvents[1]).toMatchObject({ + type: 'text_chunk', + chunk: 'hello ', + }); + expect(ch.taskEvents[2]).toMatchObject({ + type: 'tool_call', + toolCall: { toolCallId: 'tool-1' }, + }); +}); + +it('emits failed lifecycle event when prompt rejects', async () => { + vi.mocked(bridge.prompt).mockRejectedValue(new Error('boom')); + const ch = createChannel(); + + await expect(ch.handleInbound(envelope())).rejects.toThrow('boom'); + + expect(ch.taskEvents.map((event) => event.type)).toEqual([ + 'started', + 'failed', + ]); + expect(ch.taskEvents[1]).toMatchObject({ + type: 'failed', + error: 'boom', + }); +}); +``` + +For cancellation coverage, add focused assertions to existing `/cancel`, `/clear`, and `steer` tests instead of duplicating their setup: + +```ts +expect(ch.taskEvents).toContainEqual( + expect.objectContaining({ type: 'cancelled', reason: 'cancel_command' }), +); +expect(ch.taskEvents).toContainEqual( + expect.objectContaining({ type: 'cancelled', reason: 'clear' }), +); +expect(ch.taskEvents).toContainEqual( + expect.objectContaining({ type: 'cancelled', reason: 'steer' }), +); +``` + +- [ ] **Step 2: Run tests to verify failures** + +Run: + +```bash +cd packages/channels/base +npx vitest run src/ChannelBase.test.ts +``` + +Expected: fails because only `started` is emitted. + +- [ ] **Step 3: Add helper methods for lifecycle payloads** + +Add private helper: + +```ts + private lifecycleBase( + chatId: string, + sessionId: string, + messageId?: string, + ) { + return { + channelName: this.name, + chatId, + sessionId, + ...(messageId ? { messageId } : {}), + identity: this.identity, + memoryScope: this.memoryScope, + }; + } +``` + +Update `started` to spread `this.lifecycleBase(...)`. + +- [ ] **Step 4: Emit text chunk and tool call events** + +In `bridgeToolCallListener`, after `this.onToolCall(target.chatId, event);`, add: + +```ts + this.emitTaskLifecycle({ + type: 'tool_call', + channelName: this.name, + chatId: target.chatId, + sessionId: event.sessionId, + toolCall: event, + identity: this.identity, + memoryScope: this.memoryScope, + }); +``` + +In `onChunk`, after `this.onResponseChunk(...)`, add: + +```ts + this.emitTaskLifecycle({ + type: 'text_chunk', + ...this.lifecycleBase(envelope.chatId, sessionId, envelope.messageId), + chunk, + }); +``` + +- [ ] **Step 5: Emit cancellation events** + +In `/cancel`, after `active.cancelled = true;`, add: + +```ts + this.emitTaskLifecycle({ + type: 'cancelled', + ...this.lifecycleBase(active.chatId, activeSessionId, active.messageId), + reason: 'cancel_command', + }); +``` + +In `/clear`, when `active` exists before waiting, add: + +```ts + this.emitTaskLifecycle({ + type: 'cancelled', + ...this.lifecycleBase(active.chatId, id, active.messageId), + reason: 'clear', + }); +``` + +In `steer`, after `active.cancelled = true;`, add: + +```ts + this.emitTaskLifecycle({ + type: 'cancelled', + ...this.lifecycleBase(active.chatId, sessionId, active.messageId), + reason: 'steer', + }); +``` + +- [ ] **Step 6: Emit completed and failed events** + +In the prompt `try` block, after response delivery finishes and only when `!promptState.cancelled`, add: + +```ts + this.emitTaskLifecycle({ + type: 'completed', + ...this.lifecycleBase(envelope.chatId, sessionId, envelope.messageId), + }); +``` + +Convert the `try/finally` into `try/catch/finally`: + +```ts + } catch (err) { + this.emitTaskLifecycle({ + type: 'failed', + ...this.lifecycleBase(envelope.chatId, sessionId, envelope.messageId), + error: err instanceof Error ? err.message : String(err), + }); + throw err; + } finally { +``` + +- [ ] **Step 7: Run lifecycle tests** + +Run: + +```bash +cd packages/channels/base +npx vitest run src/ChannelBase.test.ts +``` + +Expected: all `ChannelBase` tests pass. + +## Task 4: Config Parsing, Exports, And Verification + +**Files:** +- Modify: `packages/cli/src/commands/channel/config-utils.ts` +- Modify: `packages/cli/src/commands/channel/config-utils.test.ts` +- Modify: `packages/channels/base/src/index.ts` +- Test: `packages/cli/src/commands/channel/config-utils.test.ts` + +- [ ] **Step 1: Write failing config parsing test** + +In `config-utils.test.ts`, add to the full config parse test or a new test: + +```ts +it('preserves channel identity and metadata-only memory scope config', async () => { + const result = await parseChannelConfig('bot', { + type: 'telegram', + token: 'tok', + identity: { + id: 'ops-agent', + displayName: 'Ops Agent', + description: 'Coordinates repository operations.', + }, + memoryScope: { + namespace: 'qwen-tag:ops', + mode: 'metadata-only', + }, + }); + + expect(result.identity).toEqual({ + id: 'ops-agent', + displayName: 'Ops Agent', + description: 'Coordinates repository operations.', + }); + expect(result.memoryScope).toEqual({ + namespace: 'qwen-tag:ops', + mode: 'metadata-only', + }); +}); +``` + +- [ ] **Step 2: Run config test to verify failure** + +Run: + +```bash +cd packages/cli +npx vitest run src/commands/channel/config-utils.test.ts +``` + +Expected: fails if parser drops the new config fields. + +- [ ] **Step 3: Preserve config fields** + +In `config-utils.ts`, add to the returned config object: + +```ts + identity: rawConfig['identity'] as ChannelConfig['identity'], + memoryScope: rawConfig['memoryScope'] as ChannelConfig['memoryScope'], +``` + +- [ ] **Step 4: Ensure public exports** + +If `types.ts` exports the new types and `index.ts` already uses `export type { ... } from './types.js';`, add the new type names there: + +```ts + ChannelIdentityConfig, + ChannelRuntimeIdentity, + ChannelMemoryScopeConfig, + ChannelRuntimeMemoryScope, + ChannelTaskLifecycleEvent, +``` + +- [ ] **Step 5: Run focused tests** + +Run: + +```bash +cd packages/channels/base +npx vitest run src/ChannelBase.test.ts src/SessionRouter.test.ts +cd ../../cli +npx vitest run src/commands/channel/config-utils.test.ts +``` + +Expected: all focused tests pass. + +- [ ] **Step 6: Run final verification** + +Run from repo root: + +```bash +npm run build +npm run typecheck +``` + +Expected: both commands pass. Existing warning-only lint output from dependency install or unrelated packages is not part of this verification. + +- [ ] **Step 7: Commit implementation** + +Review the diff and stage only intended files: + +```bash +git diff -- packages/channels/base/src/types.ts packages/channels/base/src/ChannelBase.ts packages/channels/base/src/ChannelBase.test.ts packages/channels/base/src/index.ts packages/cli/src/commands/channel/config-utils.ts packages/cli/src/commands/channel/config-utils.test.ts +git add packages/channels/base/src/types.ts packages/channels/base/src/ChannelBase.ts packages/channels/base/src/ChannelBase.test.ts packages/channels/base/src/index.ts packages/cli/src/commands/channel/config-utils.ts packages/cli/src/commands/channel/config-utils.test.ts +git commit -m "feat(channels): add identity and task lifecycle metadata" +``` + +Expected: commit contains no `package-lock.json`, generated assets, or platform adapter UI changes. diff --git a/docs/superpowers/specs/2026-07-01-channel-p0-identity-task-lifecycle-design.md b/docs/superpowers/specs/2026-07-01-channel-p0-identity-task-lifecycle-design.md new file mode 100644 index 0000000000..c3d383daa8 --- /dev/null +++ b/docs/superpowers/specs/2026-07-01-channel-p0-identity-task-lifecycle-design.md @@ -0,0 +1,272 @@ +# Channel P0 Identity And Task Lifecycle Design + +## Goal + +Implement the first P0 foundation for channel-resident multiplayer agents: +channel-scoped identity and memory-boundary metadata, plus a shared task +lifecycle hook in `@qwen-code/channel-base`. + +This intentionally does not add a Slack adapter, daemon event stream, adapter UI +changes, proactive scheduling, cross-channel context, or real core-memory path +isolation. + +## Background + +`qwen channel` already supports messaging adapters, shared sessions, sender +attribution, dispatch modes, streaming chunks, tool-call callbacks, cancellation, +and platform-specific progress surfaces such as Feishu cards. The missing P0 +product layer is a stable way to say "this channel has its own resident agent +identity" and "this prompt turn has a lifecycle adapters can observe." + +Issue #6103 tracks this focused slice. It builds on the broader qwen tag roadmap +in #5887, but keeps this PR small enough to review and ship independently. + +## Scope + +In scope: + +- Add optional channel identity metadata to `ChannelConfig`. +- Add optional memory-scope metadata to `ChannelConfig`. +- Derive safe defaults when the new config is omitted. +- Inject a concise channel boundary note into the first prompt for each agent + session, together with existing channel instructions. +- Add a protected `onTaskLifecycle(event)` hook on `ChannelBase`. +- Emit lifecycle events from the shared channel flow for prompt start, text + chunks, tool calls, cancellation, completion, and errors. +- Add focused package-local tests in `packages/channels/base`. + +Out of scope: + +- Core memory storage changes or file-path namespace isolation. +- Daemon/SSE event publication. +- Feishu, DingTalk, Telegram, WeChat, or QQ UI changes. +- New platform adapters. +- Token budgets, tool ACLs, or cross-channel context sharing. + +## Design + +### Channel Identity + +Add a small optional config object: + +```ts +export interface ChannelIdentityConfig { + id?: string; + displayName?: string; + description?: string; +} +``` + +`ChannelConfig` gains `identity?: ChannelIdentityConfig`. + +At runtime, `ChannelBase` derives: + +- `id`: `config.identity.id` or `channel:` +- `displayName`: `config.identity.displayName` or `` +- `description`: `config.identity.description`, if present + +The runtime identity is metadata only. It does not change session routing, +access control, or platform adapter behavior. + +### Memory Scope Metadata + +Add: + +```ts +export type ChannelMemoryScopeMode = 'metadata-only'; + +export interface ChannelMemoryScopeConfig { + namespace?: string; + mode?: ChannelMemoryScopeMode; +} +``` + +`ChannelConfig` gains `memoryScope?: ChannelMemoryScopeConfig`. + +At runtime, `ChannelBase` derives: + +- `namespace`: `config.memoryScope.namespace` or `channel:` +- `mode`: always `'metadata-only'` for this PR + +This is deliberately not a real core-memory namespace. It is an explicit, +inspectable boundary marker and prompt instruction so later work can wire the +same namespace into core memory paths without changing channel config shape. + +### Prompt Boundary Injection + +`ChannelBase` already prepends `config.instructions` once per session; that +behavior is unchanged. The generated boundary note below is added to the same +first-message injection only when a channel configures `identity` or +`memoryScope` (instructions-only channels keep the existing prompt shape). It +is appended after custom instructions so the boundary takes recency precedence: + +```text +Channel identity: +- id: channel:ops +- display name: Ops Bot +- description: Helps the ops group coordinate repository maintenance. + +Memory scope: +- namespace: qwen-tag:ops +- mode: metadata-only +- data from other channels must not be shared. +``` + +The exact wording should be concise and stable enough for tests, but avoid +over-promising isolation. If no description exists, omit that line. + +This note is injected once per agent session, like existing instructions +(a transient channel-memory read failure retries the whole context block on +the next turn, so consecutive turns may repeat it). When the bridge reports a +session death, the existing `instructedSessions` cleanup continues to allow +reinjection for the next session. + +For compatibility, channels with no `instructions`, `identity`, or `memoryScope` +configuration keep the existing raw prompt shape. Runtime identity and memory +metadata are still derived for lifecycle events and status commands. + +### Status Visibility + +Extend `/who` and `/status` with identity and memory metadata: + +- `/who` should include identity display name and memory namespace. +- `/status` should include the identity id and memory mode. + +Keep the output short. Do not expose absolute paths or hidden configuration. + +### Task Lifecycle Hook + +Add a discriminated union: + +```ts +export type ChannelTaskLifecycleEvent = + | { + type: 'started'; + channelName: string; + chatId: string; + sessionId: string; + messageId?: string; + identity: ChannelRuntimeIdentity; + memoryScope: ChannelRuntimeMemoryScope; + } + | { + type: 'text_chunk'; + channelName: string; + chatId: string; + sessionId: string; + messageId?: string; + chunk: string; + identity: ChannelRuntimeIdentity; + memoryScope: ChannelRuntimeMemoryScope; + } + | { + type: 'tool_call'; + channelName: string; + chatId: string; + sessionId: string; + toolCall: ToolCallEvent; + identity: ChannelRuntimeIdentity; + memoryScope: ChannelRuntimeMemoryScope; + } + | { + type: 'cancelled'; + channelName: string; + chatId: string; + sessionId: string; + messageId?: string; + reason: 'cancel_command' | 'clear' | 'steer' | 'timeout'; + identity: ChannelRuntimeIdentity; + memoryScope: ChannelRuntimeMemoryScope; + } + | { + type: 'completed'; + channelName: string; + chatId: string; + sessionId: string; + messageId?: string; + identity: ChannelRuntimeIdentity; + memoryScope: ChannelRuntimeMemoryScope; + } + | { + type: 'failed'; + channelName: string; + chatId: string; + sessionId: string; + messageId?: string; + error: string; + identity: ChannelRuntimeIdentity; + memoryScope: ChannelRuntimeMemoryScope; + }; +``` + +`ChannelBase` adds: + +```ts +protected onTaskLifecycle(_event: ChannelTaskLifecycleEvent): void {} +``` + +Default behavior is no-op. Adapters can opt in later without changing the +prompt execution path. + +### Lifecycle Emission Points + +Emit from shared `ChannelBase` flow: + +- `started`: immediately after `activePrompts.set()` and before + `onPromptStart()`. +- `text_chunk`: when the prompt's `textChunk` listener accepts a non-cancelled + chunk. +- `tool_call`: in the existing bridge tool-call listener after resolving the + session target. +- `cancelled`: when `/cancel` succeeds, when `/clear` cancels or evicts an + active prompt, and when `steer` marks the active turn cancelled. +- `completed`: after `bridge.prompt()` resolves and before or after + `onResponseComplete()`, as long as the turn was not cancelled. +- `failed`: when `bridge.prompt()` or response delivery throws. + +Lifecycle hook failures should be caught and logged to stderr. A platform +adapter's lifecycle UI must not break prompt execution or cleanup. + +## Error Handling + +- Invalid identity or memory fields are not fatal in this PR; config parsing + should preserve the existing permissive shape and only accept string fields + where explicit parsing already exists. +- Lifecycle hook exceptions are swallowed after a stderr diagnostic. +- Memory scope mode is constrained to `'metadata-only'`; omitted or unknown + config should resolve to `'metadata-only'` rather than enabling behavior that + does not exist. + +## Tests + +Focused tests in `packages/channels/base/src/ChannelBase.test.ts` should cover: + +- Default identity and memory metadata are derived from channel name. +- Custom identity and memory namespace are included in the first prompt. +- Boundary metadata is injected once per session and re-injected after + `sessionDied`. +- `/who` and `/status` include the new metadata without leaking cwd. +- `onTaskLifecycle` sees `started`, `text_chunk`, `tool_call`, `completed`. +- `onTaskLifecycle` sees `cancelled` for `/cancel`, `/clear`, and `steer`. +- `onTaskLifecycle` sees `failed` when `bridge.prompt()` rejects. +- A throwing lifecycle hook does not reject `handleInbound()`. + +Use package-local commands: + +```bash +cd packages/channels/base +npx vitest run src/ChannelBase.test.ts +``` + +Final verification before PR: + +```bash +npm run build +npm run typecheck +``` + +## Open Decisions + +None for this PR. Real core-memory namespace enforcement, daemon publication, +adapter UI, tool/data ACLs, budgets, and proactive follow-up are explicitly +future work. diff --git a/packages/channels/base/src/ChannelBase.test.ts b/packages/channels/base/src/ChannelBase.test.ts index 951e8f1aad..d278717640 100644 --- a/packages/channels/base/src/ChannelBase.test.ts +++ b/packages/channels/base/src/ChannelBase.test.ts @@ -3,7 +3,11 @@ import { EventEmitter } from 'node:events'; import { mkdtempSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import type { ChannelConfig, Envelope } from './types.js'; +import type { + ChannelConfig, + ChannelTaskLifecycleEvent, + Envelope, +} from './types.js'; import type { ChannelAgentBridge } from './ChannelAgentBridge.js'; import { ChannelBase, CLEAR_CANCEL_TIMEOUT_MS } from './ChannelBase.js'; import type { ChannelBaseOptions } from './ChannelBase.js'; @@ -16,6 +20,7 @@ class TestChannel extends ChannelBase { proactiveSupported = false; connected = false; toolCalls: Array<{ chatId: string; event: unknown }> = []; + taskEvents: ChannelTaskLifecycleEvent[] = []; promptStarts: Array<{ chatId: string; sessionId: string; @@ -27,6 +32,7 @@ class TestChannel extends ChannelBase { []; /** When set, onPromptEnd throws AFTER recording — to exercise the finally guard. */ throwOnPromptEnd = false; + responseCompleteGate?: Promise; async connect() { this.connected = true; @@ -42,6 +48,10 @@ class TestChannel extends ChannelBase { this.toolCalls.push({ chatId, event }); } + protected override onTaskLifecycle(event: ChannelTaskLifecycleEvent): void { + this.taskEvents.push(event); + } + override supportsProactiveSend(): boolean { return this.proactiveSupported; } @@ -83,6 +93,15 @@ class TestChannel extends ChannelBase { ): void { this.responseChunks.push({ chatId, chunk, sessionId }); } + + protected override async onResponseComplete( + chatId: string, + fullText: string, + sessionId: string, + ): Promise { + await this.responseCompleteGate; + await super.onResponseComplete(chatId, fullText, sessionId); + } } function createBridge(): ChannelAgentBridge { @@ -1372,6 +1391,71 @@ describe('ChannelBase', () => { expect(ch.sent[0]!.text).toContain('Channel: test-chan'); }); + it('derives default channel identity and memory metadata for task lifecycle events', async () => { + const ch = createChannel(); + + await ch.handleInbound(envelope({ messageId: 'm-1' })); + + expect(ch.taskEvents[0]).toMatchObject({ + type: 'started', + channelName: 'test-chan', + chatId: 'chat1', + sessionId: 's-1', + messageId: 'm-1', + identity: { + id: 'channel:test-chan', + displayName: 'test-chan', + }, + memoryScope: { + namespace: 'channel:test-chan', + mode: 'metadata-only', + }, + }); + }); + + it('uses configured channel identity and memory namespace in lifecycle metadata', async () => { + const ch = createChannel({ + identity: { + id: 'ops-agent', + displayName: 'Ops Agent', + description: 'Coordinates repository operations.', + }, + memoryScope: { + namespace: 'qwen-tag:ops', + mode: 'metadata-only', + }, + }); + + await ch.handleInbound(envelope()); + + expect(ch.taskEvents[0]).toMatchObject({ + identity: { + id: 'ops-agent', + displayName: 'Ops Agent', + description: 'Coordinates repository operations.', + }, + memoryScope: { + namespace: 'qwen-tag:ops', + mode: 'metadata-only', + }, + }); + }); + + it('/who and /status include channel identity and memory metadata', async () => { + const ch = createChannel({ + identity: { id: 'ops-agent', displayName: 'Ops Agent' }, + memoryScope: { namespace: 'qwen-tag:ops', mode: 'metadata-only' }, + }); + + await ch.handleInbound(envelope({ text: '/who' })); + await ch.handleInbound(envelope({ text: '/status' })); + + expect(ch.sent[0]!.text).toContain('Identity: Ops Agent'); + expect(ch.sent[0]!.text).toContain('Memory: qwen-tag:ops'); + expect(ch.sent[1]!.text).toContain('Identity: ops-agent'); + expect(ch.sent[1]!.text).toContain('Memory: metadata-only'); + }); + it('/loop add stores a job for the current channel target', async () => { const created: ChannelLoop = { id: 'job-1', @@ -2598,6 +2682,97 @@ describe('ChannelBase', () => { ); }); + it('/cancel still delivers the response when cancellation fails after the response settles', async () => { + let resolvePrompt!: (v: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + let rejectCancel!: (err: Error) => void; + const pendingCancel = new Promise((_resolve, reject) => { + rejectCancel = reject; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + (bridge.cancelSession as ReturnType).mockReturnValue( + pendingCancel, + ); + vi.spyOn(process.stderr, 'write').mockImplementation(() => true); + + const ch = createChannel(); + ch.enableCancelCommand(); + const prompt = ch.handleInbound(envelope({ text: 'long task' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + + const cancel = ch.handleInbound(envelope({ text: '/cancel' })); + await Promise.resolve(); + resolvePrompt('agent response'); + rejectCancel(new Error('session not found')); + await Promise.all([prompt, cancel]); + + expect(ch.sent).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + text: 'Failed to cancel current request.', + }), + expect.objectContaining({ text: 'agent response' }), + ]), + ); + expect(ch.sent).not.toEqual( + expect.arrayContaining([ + expect.objectContaining({ text: 'Cancelled current request.' }), + ]), + ); + }); + + it('/cancel still delivers the response when cancellation times out then fails', async () => { + vi.useFakeTimers(); + try { + let resolvePrompt!: (v: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + let rejectCancel!: (err: Error) => void; + const pendingCancel = new Promise((_resolve, reject) => { + rejectCancel = reject; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + (bridge.cancelSession as ReturnType).mockReturnValue( + pendingCancel, + ); + vi.spyOn(process.stderr, 'write').mockImplementation(() => true); + + const ch = createChannel(); + ch.enableCancelCommand(); + const prompt = ch.handleInbound(envelope({ text: 'long task' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + + const cancel = ch.handleInbound(envelope({ text: '/cancel' })); + await Promise.resolve(); + resolvePrompt('agent response'); + await vi.advanceTimersByTimeAsync(3000); + rejectCancel(new Error('session not found')); + await Promise.all([prompt, cancel]); + + expect(ch.sent).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + text: 'Failed to cancel current request.', + }), + expect.objectContaining({ text: 'agent response' }), + ]), + ); + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started' }), + expect.objectContaining({ type: 'completed' }), + ]); + } finally { + vi.useRealTimers(); + } + }); + it('/cancel retries after a failed cancellation while the prompt is still active', async () => { let resolvePrompt!: (v: string) => void; const pendingPrompt = new Promise((resolve) => { @@ -3266,6 +3441,7 @@ describe('ChannelBase', () => { // eslint-disable-next-line @typescript-eslint/no-explicit-any const firstPrompt = (bridge.prompt as any).mock.calls[0][1] as string; expect(firstPrompt).toContain('Be concise.'); + expect(firstPrompt).not.toContain('Channel identity:'); await ch.handleInbound(envelope({ text: 'second' })); // eslint-disable-next-line @typescript-eslint/no-explicit-any @@ -3273,6 +3449,110 @@ describe('ChannelBase', () => { expect(secondPrompt).not.toContain('Be concise.'); }); + it('prepends channel boundary metadata after custom instructions once per session', async () => { + const ch = createChannel({ + instructions: 'Be concise.', + identity: { + id: 'ops-agent', + displayName: 'Ops Agent', + description: 'Coordinates repository operations.', + }, + memoryScope: { + namespace: 'qwen-tag:ops', + mode: 'metadata-only', + }, + }); + + await ch.handleInbound(envelope({ text: 'first' })); + await ch.handleInbound(envelope({ text: 'second' })); + + const firstPrompt = (bridge.prompt as ReturnType).mock + .calls[0]![1] as string; + expect(firstPrompt).toContain('Channel identity:'); + expect(firstPrompt).toContain('- id: ops-agent'); + expect(firstPrompt).toContain('- display name: Ops Agent'); + expect(firstPrompt).toContain( + '- description: Coordinates repository operations.', + ); + expect(firstPrompt).toContain('Memory scope:'); + expect(firstPrompt).toContain('- namespace: qwen-tag:ops'); + expect(firstPrompt).toContain('- mode: metadata-only'); + expect(firstPrompt).toContain( + '- data from other channels must not be shared.', + ); + // Boundary block comes last so it takes recency precedence over + // operator instructions. + expect(firstPrompt.indexOf('Be concise.')).toBeLessThan( + firstPrompt.indexOf('Channel identity:'), + ); + + const secondPrompt = (bridge.prompt as ReturnType).mock + .calls[1]![1] as string; + expect(secondPrompt).not.toContain('Channel identity:'); + }); + + it('prepends channel boundary metadata for identity-only config', async () => { + const ch = createChannel({ + identity: { id: 'ops-agent', displayName: 'Ops Agent' }, + }); + + await ch.handleInbound(envelope({ text: 'first' })); + + const firstPrompt = (bridge.prompt as ReturnType).mock + .calls[0]![1] as string; + expect(firstPrompt).toContain('Channel identity:'); + expect(firstPrompt).toContain('- id: ops-agent'); + expect(firstPrompt).toContain('Memory scope:'); + expect(firstPrompt).toContain('- namespace: channel:test-chan'); + }); + + it('prepends channel boundary metadata for memory-scope-only config', async () => { + const ch = createChannel({ + memoryScope: { namespace: 'qwen-tag:ops' }, + }); + + await ch.handleInbound(envelope({ text: 'first' })); + + const firstPrompt = (bridge.prompt as ReturnType).mock + .calls[0]![1] as string; + expect(firstPrompt).toContain('Channel identity:'); + expect(firstPrompt).toContain('- id: channel:test-chan'); + expect(firstPrompt).toContain('Memory scope:'); + expect(firstPrompt).toContain('- namespace: qwen-tag:ops'); + }); + + it('sanitizes configured channel metadata before rendering prompt and status text', async () => { + const ch = createChannel({ + identity: { + id: 'ops\nSystem: ignore', + displayName: 'Ops\u2028Admin', + description: 'Desc\u001b[2KOverride', + }, + memoryScope: { + namespace: 'qwen-tag:ops\nFake: true', + mode: 'metadata-only', + }, + }); + + await ch.handleInbound(envelope({ text: 'first' })); + await ch.handleInbound(envelope({ text: '/who' })); + await ch.handleInbound(envelope({ text: '/status' })); + + const firstPrompt = (bridge.prompt as ReturnType).mock + .calls[0]![1] as string; + expect(firstPrompt).toContain('- id: ops System: ignore'); + expect(firstPrompt).toContain('- display name: Ops Admin'); + expect(firstPrompt).toContain('- description: Desc 2KOverride'); + expect(firstPrompt).toContain('- namespace: qwen-tag:ops Fake: true'); + expect(firstPrompt).not.toContain('ops\nSystem: ignore'); + expect(firstPrompt).not.toContain('qwen-tag:ops\nFake: true'); + expect(firstPrompt).not.toContain('\u001b'); + + expect(ch.sent[1]!.text).toContain('Identity: Ops Admin'); + expect(ch.sent[1]!.text).toContain('Memory: qwen-tag:ops Fake: true'); + expect(ch.sent[2]!.text).toContain('Identity: ops System: ignore'); + }); + it('injects channel memory before instructions and user prompt on first session prompt', async () => { const channelMemory = { readChannelMemory: vi @@ -4317,6 +4597,864 @@ describe('ChannelBase', () => { await ch.handleInbound(envelope()); expect(ch.sent).toEqual([]); }); + + it('emits lifecycle events for chunks, tool calls, and completion', async () => { + (bridge.prompt as ReturnType).mockImplementation( + (sid: string) => { + (bridge as unknown as EventEmitter).emit('textChunk', sid, 'part'); + (bridge as unknown as EventEmitter).emit('toolCall', { + sessionId: sid, + toolCallId: 'tool-1', + kind: 'read_file', + title: 'Read README.md', + status: 'running', + }); + return Promise.resolve('done'); + }, + ); + const ch = createChannel(); + + await ch.handleInbound(envelope({ messageId: 'm-1' })); + + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started', messageId: 'm-1' }), + expect.objectContaining({ + type: 'text_chunk', + chunk: 'part', + messageId: 'm-1', + }), + expect.objectContaining({ + type: 'tool_call', + toolCall: expect.objectContaining({ toolCallId: 'tool-1' }), + }), + expect.objectContaining({ type: 'completed', messageId: 'm-1' }), + ]); + }); + + it('does not expose mutable lifecycle metadata references', async () => { + (bridge.prompt as ReturnType).mockResolvedValue('done'); + const ch = createChannel({ + identity: { id: 'team-bot', displayName: 'Team Bot' }, + memoryScope: { namespace: 'team-chat' }, + }); + + await ch.handleInbound(envelope()); + const started = ch.taskEvents.find((event) => event.type === 'started'); + expect(started).toBeDefined(); + expect(() => { + started!.identity.displayName = 'mutated'; + }).toThrow(TypeError); + expect(() => { + started!.memoryScope.namespace = 'mutated-memory'; + }).toThrow(TypeError); + + await ch.handleInbound(envelope({ text: '/who' })); + + expect(ch.sent.at(-1)!.text).toContain('Identity: Team Bot'); + expect(ch.sent.at(-1)!.text).toContain('Memory: team-chat'); + }); + + it('strips raw tool input from lifecycle events while preserving adapter tool calls', async () => { + (bridge.prompt as ReturnType).mockImplementation( + (sid: string) => { + (bridge as unknown as EventEmitter).emit('toolCall', { + sessionId: sid, + toolCallId: 'tool-1', + kind: `run_shell_command\n${'k'.repeat(100)}`, + title: `Run shell command: echo $SECRET\n${'x'.repeat(100)}`, + status: `running\n${'s'.repeat(100)}`, + rawInput: { command: 'echo $SECRET' }, + }); + return Promise.resolve('done'); + }, + ); + const ch = createChannel(); + + await ch.handleInbound(envelope()); + + const lifecycleToolCall = ch.taskEvents.find( + (event) => event.type === 'tool_call', + ); + expect(lifecycleToolCall).toMatchObject({ + type: 'tool_call', + toolCall: expect.objectContaining({ + toolCallId: 'tool-1', + }), + }); + expect(lifecycleToolCall!.toolCall).not.toHaveProperty('rawInput'); + expect(lifecycleToolCall!.toolCall.kind).not.toContain('\n'); + expect(lifecycleToolCall!.toolCall.status).not.toContain('\n'); + expect( + Array.from(lifecycleToolCall!.toolCall.kind).length, + ).toBeLessThanOrEqual(21); + expect( + Array.from(lifecycleToolCall!.toolCall.status).length, + ).toBeLessThanOrEqual(21); + expect(lifecycleToolCall!.toolCall.title).not.toContain('\n'); + expect( + Array.from(lifecycleToolCall!.toolCall.title).length, + ).toBeLessThanOrEqual(81); + expect(ch.toolCalls[0]!.event).toMatchObject({ + rawInput: { command: 'echo $SECRET' }, + }); + }); + + it('dispatches shared-router tool calls through the active prompt context', async () => { + let resolveSecond!: (value: string) => void; + const pendingSecond = new Promise((resolve) => { + resolveSecond = resolve; + }); + (bridge.prompt as ReturnType) + .mockResolvedValueOnce('first') + .mockReturnValueOnce(pendingSecond); + const ch = createChannel({ sessionScope: 'single' }); + + await ch.handleInbound( + envelope({ chatId: 'first-chat', senderId: 'alice' }), + ); + const secondPrompt = ch.handleInbound( + envelope({ + chatId: 'second-chat', + senderId: 'bob', + messageId: 'second-message', + }), + ); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(2)); + const sessionId = (bridge.prompt as ReturnType).mock + .calls[1]![0] as string; + const toolCall = { + sessionId, + toolCallId: 'tool-shared', + kind: 'read_file', + title: 'Read README.md', + status: 'running', + rawInput: { path: 'README.md' }, + }; + + ch.dispatchToolCall(toolCall); + + expect(ch.toolCalls).toEqual([ + { chatId: 'second-chat', event: toolCall }, + ]); + expect(ch.taskEvents).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: 'tool_call', + chatId: 'second-chat', + messageId: 'second-message', + toolCall: expect.objectContaining({ toolCallId: 'tool-shared' }), + }), + ]), + ); + const lifecycleToolCall = ch.taskEvents.find( + (event) => event.type === 'tool_call', + ); + expect(lifecycleToolCall!.toolCall).not.toHaveProperty('rawInput'); + + resolveSecond('second response'); + await secondPrompt; + }); + + it('emits failed lifecycle event when prompting rejects', async () => { + (bridge.prompt as ReturnType).mockRejectedValue( + new Error('agent boom'), + ); + const ch = createChannel(); + + await expect(ch.handleInbound(envelope())).rejects.toThrow('agent boom'); + + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started' }), + expect.objectContaining({ + type: 'failed', + error: 'agent boom', + phase: 'agent', + }), + ]); + }); + + it('contains a throwing onTaskLifecycle hook and logs it', async () => { + class ThrowingChannel extends TestChannel { + protected override onTaskLifecycle( + event: ChannelTaskLifecycleEvent, + ): void { + super.onTaskLifecycle(event); + throw new Error('hook boom'); + } + } + const stderr = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + try { + const ch = new ThrowingChannel('test-chan', defaultConfig(), bridge); + await ch.handleInbound(envelope()); + + expect(ch.sent).toEqual([{ chatId: 'chat1', text: 'agent response' }]); + expect(ch.taskEvents.map((event) => event.type)).toEqual([ + 'started', + 'completed', + ]); + expect(stderr).toHaveBeenCalledWith( + expect.stringContaining( + 'onTaskLifecycle threw for started session s-1: hook boom', + ), + ); + } finally { + stderr.mockRestore(); + } + }); + + it('logs turn errors that arrive after cancellation', async () => { + let rejectPrompt!: (error: Error) => void; + (bridge.prompt as ReturnType).mockReturnValue( + new Promise((_resolve, reject) => { + rejectPrompt = reject; + }), + ); + const stderr = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + const ch = createChannel(); + ch.enableCancelCommand(); + try { + const prompt = ch.handleInbound(envelope({ messageId: 'm-9' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + await ch.handleInbound(envelope({ text: '/cancel' })); + rejectPrompt(new Error('bridge crashed')); + + await expect(prompt).rejects.toThrow('bridge crashed'); + expect( + ch.taskEvents.filter((event) => event.type === 'failed'), + ).toEqual([]); + expect(stderr).toHaveBeenCalledWith( + expect.stringContaining( + '[test-chan] turn m-9 threw after cancellation for session s-1: bridge crashed', + ), + ); + } finally { + stderr.mockRestore(); + } + }); + + it('sanitizes failed lifecycle errors', async () => { + (bridge.prompt as ReturnType).mockRejectedValue( + new Error('agent boom\nsecret second line'), + ); + const ch = createChannel(); + + await expect(ch.handleInbound(envelope())).rejects.toThrow('agent boom'); + + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started' }), + expect.objectContaining({ + type: 'failed', + error: 'agent boom\\nsecret second line', + }), + ]); + }); + + it('logs async lifecycle hook errors without disrupting the prompt flow', async () => { + (bridge.prompt as ReturnType).mockResolvedValue('ok'); + const stderr = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + const ch = createChannel(); + vi.spyOn( + ch as unknown as { + onTaskLifecycle: ( + event: ChannelTaskLifecycleEvent, + ) => void | Promise; + }, + 'onTaskLifecycle', + ).mockImplementation((event) => { + if (event.type === 'started') { + return Promise.reject(new Error('async hook failed')); + } + return undefined; + }); + + try { + await ch.handleInbound(envelope()); + + expect(ch.sent).toEqual([{ chatId: 'chat1', text: 'ok' }]); + await vi.waitFor(() => + expect(stderr).toHaveBeenCalledWith( + expect.stringContaining( + 'onTaskLifecycle threw for started session s-1: async hook failed', + ), + ), + ); + } finally { + stderr.mockRestore(); + } + }); + + it('emits cancellation lifecycle event for /cancel', async () => { + let resolvePrompt!: (value: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + const ch = createChannel(); + ch.enableCancelCommand(); + + const prompt = ch.handleInbound(envelope({ messageId: 'm-cancel' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + await ch.handleInbound(envelope({ text: '/cancel' })); + resolvePrompt('late'); + await prompt; + + expect(ch.taskEvents).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: 'cancelled', + reason: 'cancel_command', + messageId: 'm-cancel', + }), + ]), + ); + expect(ch.taskEvents).not.toEqual( + expect.arrayContaining([ + expect.objectContaining({ type: 'completed' }), + ]), + ); + }); + + it('suppresses lifecycle activity while cancel command is still awaiting bridge cancellation', async () => { + let resolvePrompt!: (value: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + let resolveCancel!: () => void; + const pendingCancel = new Promise((resolve) => { + resolveCancel = resolve; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + (bridge.cancelSession as ReturnType).mockReturnValue( + pendingCancel, + ); + const ch = createChannel(); + ch.enableCancelCommand(); + + const prompt = ch.handleInbound(envelope({ messageId: 'm-cancel' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + const sessionId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + + const cancel = ch.handleInbound(envelope({ text: '/cancel' })); + await Promise.resolve(); + (bridge as unknown as EventEmitter).emit( + 'textChunk', + sessionId, + 'late part', + ); + resolvePrompt('late response'); + await Promise.resolve(); + const eventTypes = ch.taskEvents.map((event) => event.type); + expect(eventTypes).not.toContain('text_chunk'); + expect(eventTypes).not.toContain('completed'); + expect(ch.responseChunks).toEqual([]); + resolveCancel(); + await Promise.all([prompt, cancel]); + + expect(ch.sent).toEqual([ + { chatId: 'chat1', text: 'Cancelled current request.' }, + ]); + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started', messageId: 'm-cancel' }), + expect.objectContaining({ + type: 'cancelled', + reason: 'cancel_command', + messageId: 'm-cancel', + }), + ]); + }); + + it('does not emit tool call lifecycle events while cancellation is pending', async () => { + let resolvePrompt!: (value: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + let resolveCancel!: () => void; + const pendingCancel = new Promise((resolve) => { + resolveCancel = resolve; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + (bridge.cancelSession as ReturnType).mockReturnValue( + pendingCancel, + ); + const ch = createChannel(); + ch.enableCancelCommand(); + + const prompt = ch.handleInbound(envelope({ messageId: 'm-cancel' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + const sessionId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + + const cancel = ch.handleInbound(envelope({ text: '/cancel' })); + await Promise.resolve(); + (bridge as unknown as EventEmitter).emit('toolCall', { + sessionId, + toolCallId: 'tool-pending-cancel', + kind: 'read_file', + title: 'Read README.md', + status: 'running', + }); + + expect(ch.toolCalls).toEqual([ + { + chatId: 'chat1', + event: expect.objectContaining({ + toolCallId: 'tool-pending-cancel', + }), + }, + ]); + expect(ch.taskEvents).not.toEqual( + expect.arrayContaining([ + expect.objectContaining({ type: 'tool_call' }), + ]), + ); + resolveCancel(); + await cancel; + resolvePrompt('late'); + await prompt; + }); + + it('reports cancel failure once response delivery has started', async () => { + let releaseDelivery!: () => void; + const deliveryGate = new Promise((resolve) => { + releaseDelivery = resolve; + }); + (bridge.prompt as ReturnType).mockResolvedValue('done'); + const ch = createChannel(); + ch.enableCancelCommand(); + ch.responseCompleteGate = deliveryGate; + + const prompt = ch.handleInbound(envelope({ messageId: 'm-cancel' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + // The turn is now blocked inside delivery — a cancel can no longer + // suppress the output, so it must fail honestly instead of emitting a + // cancelled event for a response the user will receive. + await ch.handleInbound(envelope({ text: '/cancel' })); + + expect(bridge.cancelSession).not.toHaveBeenCalled(); + expect(ch.sent).toContainEqual({ + chatId: 'chat1', + text: 'Failed to cancel current request.', + }); + + releaseDelivery(); + await prompt; + expect(ch.taskEvents.map((event) => event.type)).toEqual([ + 'started', + 'completed', + ]); + }); + + it('delivers completion when cancellation outlives the reconciliation timeout', async () => { + let resolvePrompt!: (value: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + let resolveCancel!: () => void; + const pendingCancel = new Promise((resolve) => { + resolveCancel = resolve; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + (bridge.cancelSession as ReturnType).mockReturnValue( + pendingCancel, + ); + const ch = createChannel(); + ch.enableCancelCommand(); + + const prompt = ch.handleInbound(envelope({ messageId: 'm-cancel' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + const cancel = ch.handleInbound(envelope({ text: '/cancel' })); + await Promise.resolve(); + + resolvePrompt('late response'); + await new Promise((resolve) => + setTimeout(resolve, CLEAR_CANCEL_TIMEOUT_MS + 20), + ); + await prompt; + + expect(ch.sent).toEqual([{ chatId: 'chat1', text: 'late response' }]); + expect(ch.taskEvents.map((event) => event.type)).toEqual([ + 'started', + 'completed', + ]); + + resolveCancel(); + await cancel; + + expect(ch.sent).toEqual([ + { chatId: 'chat1', text: 'late response' }, + { chatId: 'chat1', text: 'Failed to cancel current request.' }, + ]); + expect(ch.taskEvents.map((event) => event.type)).toEqual([ + 'started', + 'completed', + ]); + }, 8000); + + it('emits one cancellation lifecycle event for repeated /cancel commands', async () => { + let resolvePrompt!: (value: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + let resolveCancel!: () => void; + const pendingCancel = new Promise((resolve) => { + resolveCancel = resolve; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + (bridge.cancelSession as ReturnType).mockReturnValue( + pendingCancel, + ); + const ch = createChannel(); + ch.enableCancelCommand(); + + const prompt = ch.handleInbound(envelope({ messageId: 'm-cancel' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + const firstCancel = ch.handleInbound(envelope({ text: '/cancel' })); + const secondCancel = ch.handleInbound(envelope({ text: '/cancel' })); + await Promise.resolve(); + + expect(bridge.cancelSession).toHaveBeenCalledTimes(1); + resolveCancel(); + await firstCancel; + await secondCancel; + resolvePrompt('late'); + await prompt; + + const cancelEvents = ch.taskEvents.filter( + (event) => event.type === 'cancelled', + ); + expect(cancelEvents).toHaveLength(1); + expect(cancelEvents[0]).toMatchObject({ + reason: 'cancel_command', + messageId: 'm-cancel', + }); + }); + + it('does not emit failed after a cancelled prompt rejects', async () => { + let rejectPrompt!: (error: Error) => void; + const pendingPrompt = new Promise((_resolve, reject) => { + rejectPrompt = reject; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + const ch = createChannel(); + ch.enableCancelCommand(); + + const prompt = ch.handleInbound(envelope({ messageId: 'm-cancel' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + await ch.handleInbound(envelope({ text: '/cancel' })); + rejectPrompt(new Error('bridge cancelled')); + await expect(prompt).rejects.toThrow('bridge cancelled'); + + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ + type: 'started', + messageId: 'm-cancel', + }), + expect.objectContaining({ + type: 'cancelled', + reason: 'cancel_command', + messageId: 'm-cancel', + }), + ]); + }); + + it('does not emit tool call lifecycle events after cancellation', async () => { + let resolvePrompt!: (value: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + const ch = createChannel(); + ch.enableCancelCommand(); + + const prompt = ch.handleInbound(envelope({ messageId: 'm-cancel' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + const sessionId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + + await ch.handleInbound(envelope({ text: '/cancel' })); + (bridge as unknown as EventEmitter).emit('toolCall', { + sessionId, + toolCallId: 'tool-after-cancel', + name: 'read_file', + args: { path: 'README.md' }, + }); + resolvePrompt('late'); + await prompt; + + expect(ch.toolCalls).toEqual([ + { + chatId: 'chat1', + event: expect.objectContaining({ + toolCallId: 'tool-after-cancel', + }), + }, + ]); + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ + type: 'started', + messageId: 'm-cancel', + }), + expect.objectContaining({ + type: 'cancelled', + reason: 'cancel_command', + messageId: 'm-cancel', + }), + ]); + }); + + it('emits cancellation lifecycle event for /clear', async () => { + let resolvePrompt!: (value: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + (bridge.cancelSession as ReturnType).mockImplementation( + () => { + resolvePrompt('late'); + return Promise.resolve(); + }, + ); + const ch = createChannel(); + + const prompt = ch.handleInbound(envelope({ messageId: 'm-clear' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + await ch.handleInbound(envelope({ text: '/clear' })); + await prompt; + + expect(ch.taskEvents).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: 'cancelled', + reason: 'clear', + messageId: 'm-clear', + }), + ]), + ); + }); + + it('emits /clear cancellation lifecycle before prompt end cleanup', async () => { + let resolvePrompt!: (value: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + (bridge.cancelSession as ReturnType).mockImplementation( + () => { + order.push('cancelSession'); + resolvePrompt('late'); + return Promise.resolve(); + }, + ); + const ch = createChannel(); + const order: string[] = []; + vi.spyOn( + ch as unknown as { + onTaskLifecycle: (event: ChannelTaskLifecycleEvent) => void; + }, + 'onTaskLifecycle', + ).mockImplementation((event) => { + if (event.type === 'cancelled') { + order.push('cancelled'); + } + }); + vi.spyOn( + ch as unknown as { + onPromptEnd: ( + chatId: string, + sessionId: string, + messageId?: string, + ) => void; + }, + 'onPromptEnd', + ).mockImplementation(() => { + order.push('end'); + }); + + const prompt = ch.handleInbound(envelope({ messageId: 'm-clear' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + await ch.handleInbound(envelope({ text: '/clear' })); + await prompt; + + expect(order).toEqual(['cancelSession', 'cancelled', 'end']); + }); + + it('does not emit a second cancellation lifecycle event when /clear follows /cancel', async () => { + let resolvePrompt!: (value: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + (bridge.cancelSession as ReturnType).mockResolvedValue( + undefined, + ); + const ch = createChannel(); + ch.enableCancelCommand(); + + const prompt = ch.handleInbound( + envelope({ messageId: 'm-cancel-clear' }), + ); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + await ch.handleInbound(envelope({ text: '/cancel' })); + const clear = ch.handleInbound(envelope({ text: '/clear' })); + resolvePrompt('late'); + await prompt; + await clear; + + const cancelEvents = ch.taskEvents.filter( + (event) => event.type === 'cancelled', + ); + expect(cancelEvents).toHaveLength(1); + expect(cancelEvents[0]).toMatchObject({ + reason: 'cancel_command', + messageId: 'm-cancel-clear', + }); + }); + + it('emits cancellation lifecycle event for steer', async () => { + let resolveFirst!: (value: string) => void; + const firstPrompt = new Promise((resolve) => { + resolveFirst = resolve; + }); + (bridge.prompt as ReturnType) + .mockReturnValueOnce(firstPrompt) + .mockResolvedValueOnce('second'); + (bridge.cancelSession as ReturnType).mockImplementation( + () => { + resolveFirst('late'); + return Promise.resolve(); + }, + ); + const ch = createChannel(); + + const first = ch.handleInbound(envelope({ messageId: 'm-steer' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + const second = ch.handleInbound(envelope({ text: 'replacement' })); + await first; + await second; + + expect(ch.taskEvents).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: 'cancelled', + reason: 'steer', + messageId: 'm-steer', + }), + ]), + ); + }); + + it('stops active streaming before emitting steer cancellation lifecycle', async () => { + let resolveFirst!: (value: string) => void; + const firstPrompt = new Promise((resolve) => { + resolveFirst = resolve; + }); + (bridge.prompt as ReturnType) + .mockReturnValueOnce(firstPrompt) + .mockResolvedValueOnce('second'); + (bridge.cancelSession as ReturnType).mockImplementation( + () => { + resolveFirst('late'); + return Promise.resolve(); + }, + ); + const ch = createChannel(); + const order: string[] = []; + vi.spyOn( + ch as unknown as { + stopActiveStreaming: ( + active: unknown, + sessionId: string, + reason: string, + ) => void; + }, + 'stopActiveStreaming', + ).mockImplementation(() => { + order.push('stop'); + }); + vi.spyOn( + ch as unknown as { + onTaskLifecycle: (event: ChannelTaskLifecycleEvent) => void; + }, + 'onTaskLifecycle', + ).mockImplementation((event) => { + if (event.type === 'cancelled') { + order.push('cancelled'); + } + }); + + const first = ch.handleInbound(envelope({ messageId: 'm-steer' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + const second = ch.handleInbound(envelope({ text: 'replacement' })); + await first; + await second; + + expect(order).toEqual(['stop', 'cancelled']); + }); + + it('emits one cancellation lifecycle event for repeated steer messages before the active turn settles', async () => { + let resolveFirst!: (value: string) => void; + const firstPrompt = new Promise((resolve) => { + resolveFirst = resolve; + }); + (bridge.prompt as ReturnType) + .mockReturnValueOnce(firstPrompt) + .mockResolvedValueOnce('second') + .mockResolvedValueOnce('third'); + (bridge.cancelSession as ReturnType).mockResolvedValue( + undefined, + ); + const ch = createChannel(); + + const first = ch.handleInbound(envelope({ messageId: 'm-steer' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + const second = ch.handleInbound(envelope({ text: 'replacement one' })); + const third = ch.handleInbound(envelope({ text: 'replacement two' })); + for (let i = 0; i < 10; i++) { + await Promise.resolve(); + } + + expect(bridge.cancelSession).toHaveBeenCalledTimes(1); + resolveFirst('late'); + await first; + await second; + await third; + + const cancelEvents = ch.taskEvents.filter( + (event) => event.type === 'cancelled', + ); + expect(cancelEvents).toHaveLength(1); + expect(cancelEvents[0]).toMatchObject({ + reason: 'steer', + messageId: 'm-steer', + }); + }); }); describe('block streaming', () => { @@ -4419,6 +5557,100 @@ describe('ChannelBase', () => { vi.useRealTimers(); } }); + + it('keeps block-streaming chunks emitted while a failed cancel is pending', async () => { + let resolvePrompt!: (v: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + let rejectCancel!: (err: Error) => void; + const pendingCancel = new Promise((_resolve, reject) => { + rejectCancel = reject; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + (bridge.cancelSession as ReturnType).mockReturnValue( + pendingCancel, + ); + vi.spyOn(process.stderr, 'write').mockImplementation(() => true); + + const ch = createChannel({ + blockStreaming: 'on', + blockStreamingChunk: { minChars: 5, maxChars: 1000 }, + blockStreamingCoalesce: { idleMs: 500 }, + }); + ch.enableCancelCommand(); + const prompt = ch.handleInbound(envelope({ text: 'long task' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + + (bridge as unknown as EventEmitter).emit('textChunk', 's-1', 'before '); + const cancel = ch.handleInbound(envelope({ text: '/cancel' })); + await Promise.resolve(); + (bridge as unknown as EventEmitter).emit('textChunk', 's-1', 'during '); + rejectCancel(new Error('session not found')); + await cancel; + (bridge as unknown as EventEmitter).emit('textChunk', 's-1', 'after'); + resolvePrompt('before during after'); + await prompt; + + expect(ch.sent.map((message) => message.text).join('\n')).toContain( + 'before during after', + ); + expect(ch.responseChunks.map((entry) => entry.chunk)).toEqual([ + 'before ', + 'during ', + 'after', + ]); + }); + + it('never sends held block-streaming chunks when the pending cancel succeeds', async () => { + let resolvePrompt!: (v: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + let resolveCancel!: () => void; + const pendingCancel = new Promise((resolve) => { + resolveCancel = resolve; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + (bridge.cancelSession as ReturnType).mockReturnValue( + pendingCancel, + ); + + const ch = createChannel({ + blockStreaming: 'on', + blockStreamingChunk: { minChars: 5, maxChars: 10 }, + blockStreamingCoalesce: { idleMs: 500 }, + }); + ch.enableCancelCommand(); + const prompt = ch.handleInbound(envelope({ text: 'long task' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + + const cancel = ch.handleInbound(envelope({ text: '/cancel' })); + await Promise.resolve(); + // Far past every send threshold — pushing this into the BlockStreamer + // during the pending window would emit a block the cancel can't recall. + (bridge as unknown as EventEmitter).emit( + 'textChunk', + 's-1', + 'paragraph one.\n\nparagraph two.\n\n', + ); + resolveCancel(); + await cancel; + resolvePrompt('paragraph one.\n\nparagraph two.'); + await prompt; + + expect(ch.sent).toEqual([ + { chatId: 'chat1', text: 'Cancelled current request.' }, + ]); + expect( + ch.taskEvents.filter((event) => event.type === 'text_chunk'), + ).toEqual([]); + expect(ch.responseChunks).toEqual([]); + }); }); describe('pairing flow', () => { @@ -6352,6 +7584,77 @@ describe('ChannelBase', () => { ]); }); + it('prepends channel boundary metadata to first loop prompt in a session', async () => { + const ch = createChannel({ + instructions: 'Reply briefly.', + identity: { + id: 'channel:test', + displayName: 'Test Channel', + }, + memoryScope: { + namespace: 'memory:test', + }, + }); + ch.proactiveSupported = true; + + await ch.runLoopPrompt({ + id: 'job-1', + channelName: 'test-chan', + target: { + channelName: 'test-chan', + senderId: 'alice', + chatId: 'chat1', + isGroup: false, + }, + cwd: '/tmp', + cron: '0 9 * * *', + prompt: 'post summary', + label: 'daily summary', + recurring: true, + enabled: true, + createdBy: 'Alice', + createdAt: '2026-06-30T01:00:00.000Z', + consecutiveFailures: 0, + runCount: 0, + }); + await ch.runLoopPrompt({ + id: 'job-2', + channelName: 'test-chan', + target: { + channelName: 'test-chan', + senderId: 'alice', + chatId: 'chat1', + isGroup: false, + }, + cwd: '/tmp', + cron: '0 9 * * *', + prompt: 'post summary again', + label: 'daily summary', + recurring: true, + enabled: true, + createdBy: 'Alice', + createdAt: '2026-06-30T01:00:00.000Z', + consecutiveFailures: 0, + runCount: 1, + }); + + const promptText = (bridge.prompt as ReturnType).mock + .calls[0]![1] as string; + expect(promptText).toContain('Channel identity:\n- id: channel:test'); + expect(promptText).toContain('- display name: Test Channel'); + expect(promptText).toContain('- namespace: memory:test'); + expect(promptText).toContain('Reply briefly.'); + expect(promptText).toContain( + '[Loop "daily summary" created by Alice]\n\npost summary', + ); + const secondPromptText = (bridge.prompt as ReturnType).mock + .calls[1]![1] as string; + expect(secondPromptText).not.toContain('Channel identity:'); + expect(secondPromptText).toContain( + '[Loop "daily summary" created by Alice]\n\npost summary again', + ); + }); + it('injects channel memory before instructions for first loop prompt in a session', async () => { const channelMemory = { readChannelMemory: vi.fn().mockResolvedValue('Use staging.\n'), @@ -6460,6 +7763,7 @@ describe('ChannelBase', () => { '[Loop "daily summary" created by Alice] Scheduled task running unattended: no one is present to answer questions, and your final response is delivered to this chat automatically — do whatever work the task requires, then put the result in your final response instead of trying to deliver it to this chat yourself.\n\npost summary', ].join('\n\n'), ); + stderr.mockRestore(); }); it('drops a loop prompt cleared during a slow memory read', async () => { @@ -6512,6 +7816,224 @@ describe('ChannelBase', () => { expect(bridge.prompt).not.toHaveBeenCalled(); }); + it('emits lifecycle events for loop chunks and completion', async () => { + (bridge.prompt as ReturnType).mockImplementation( + (sid: string) => { + (bridge as unknown as EventEmitter).emit('textChunk', sid, 'part'); + return Promise.resolve('loop response'); + }, + ); + const ch = createChannel(); + ch.proactiveSupported = true; + + await ch.runLoopPrompt({ + id: 'job-1', + channelName: 'test-chan', + target: { + channelName: 'test-chan', + senderId: 'alice', + chatId: 'chat1', + isGroup: false, + }, + cwd: '/tmp', + cron: '0 9 * * *', + prompt: 'post summary', + label: 'daily summary', + recurring: true, + enabled: true, + createdBy: 'Alice', + createdAt: '2026-06-30T01:00:00.000Z', + consecutiveFailures: 0, + runCount: 0, + }); + + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started', messageId: 'job-1' }), + expect.objectContaining({ + type: 'text_chunk', + chunk: 'part', + messageId: 'job-1', + }), + expect.objectContaining({ type: 'completed', messageId: 'job-1' }), + ]); + }); + + it('suppresses loop chunks while cancellation is pending', async () => { + let resolvePrompt!: (value: string) => void; + const pendingPrompt = new Promise((resolve) => { + resolvePrompt = resolve; + }); + let resolveCancel!: () => void; + const pendingCancel = new Promise((resolve) => { + resolveCancel = resolve; + }); + (bridge.prompt as ReturnType).mockReturnValue( + pendingPrompt, + ); + (bridge.cancelSession as ReturnType).mockReturnValue( + pendingCancel, + ); + const ch = createChannel(); + ch.enableCancelCommand(); + ch.proactiveSupported = true; + + const loopRun = ch.runLoopPrompt({ + id: 'job-1', + channelName: 'test-chan', + target: { + channelName: 'test-chan', + senderId: 'alice', + chatId: 'chat1', + isGroup: false, + }, + cwd: '/tmp', + cron: '0 9 * * *', + prompt: 'post summary', + label: 'daily summary', + recurring: true, + enabled: true, + createdBy: 'Alice', + createdAt: '2026-06-30T01:00:00.000Z', + consecutiveFailures: 0, + runCount: 0, + }); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + const sessionId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + + const cancel = ch.handleInbound( + envelope({ senderId: 'alice', text: '/cancel' }), + ); + await Promise.resolve(); + (bridge as unknown as EventEmitter).emit( + 'textChunk', + sessionId, + 'late loop part', + ); + + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started', messageId: 'job-1' }), + ]); + expect(ch.responseChunks).toEqual([]); + resolveCancel(); + await cancel; + resolvePrompt('loop response'); + await expect(loopRun).rejects.toThrow('loop cancelled before delivery'); + // Held chunk is discarded on a successful cancel — no text_chunk event. + expect( + ch.taskEvents.filter((event) => event.type === 'text_chunk'), + ).toEqual([]); + }); + + it('emits a terminal lifecycle event when a loop is disabled after the agent response', async () => { + const shouldContinue = vi + .fn() + .mockResolvedValueOnce(true) + .mockResolvedValueOnce(false); + (bridge.prompt as ReturnType).mockResolvedValue( + 'loop response', + ); + const ch = createChannel(); + ch.proactiveSupported = true; + + await expect( + ch.runLoopPrompt( + { + id: 'job-1', + channelName: 'test-chan', + target: { + channelName: 'test-chan', + senderId: 'alice', + chatId: 'chat1', + isGroup: false, + }, + cwd: '/tmp', + cron: '0 9 * * *', + prompt: 'post summary', + label: 'daily summary', + recurring: true, + enabled: true, + createdBy: 'Alice', + createdAt: '2026-06-30T01:00:00.000Z', + consecutiveFailures: 0, + runCount: 0, + }, + { shouldContinue }, + ), + ).rejects.toThrow('loop dropped before delivery'); + + expect(ch.proactive).toEqual([]); + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started', messageId: 'job-1' }), + expect.objectContaining({ + type: 'cancelled', + messageId: 'job-1', + reason: 'dropped', + }), + ]); + }); + + it('logs loop prompt errors that arrive after cancellation', async () => { + let rejectPrompt!: (error: Error) => void; + (bridge.prompt as ReturnType).mockReturnValue( + new Promise((_resolve, reject) => { + rejectPrompt = reject; + }), + ); + const stderr = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + const ch = createChannel(); + ch.enableCancelCommand(); + ch.proactiveSupported = true; + + try { + const loopRun = ch.runLoopPrompt({ + id: 'job-1', + channelName: 'test-chan', + target: { + channelName: 'test-chan', + senderId: 'alice', + chatId: 'chat1', + isGroup: false, + }, + cwd: '/tmp', + cron: '0 9 * * *', + prompt: 'post summary', + label: 'daily summary', + recurring: true, + enabled: true, + createdBy: 'Alice', + createdAt: '2026-06-30T01:00:00.000Z', + consecutiveFailures: 0, + runCount: 0, + }); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + + await ch.handleInbound( + envelope({ text: '/cancel', senderId: 'alice' }), + ); + rejectPrompt(new Error('bridge crashed')); + + await expect(loopRun).rejects.toThrow('bridge crashed'); + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started', messageId: 'job-1' }), + expect.objectContaining({ + type: 'cancelled', + messageId: 'job-1', + reason: 'cancel_command', + }), + ]); + expect(stderr).toHaveBeenCalledWith( + expect.stringContaining( + '[test-chan] loop job-1 threw after cancellation for session s-1: bridge crashed', + ), + ); + } finally { + stderr.mockRestore(); + } + }); + it('disables single-scope loop prompts before they reach the agent', async () => { const disable = vi.fn().mockResolvedValue(true); const ch = createChannel( @@ -6685,6 +8207,14 @@ describe('ChannelBase', () => { } ).activePrompts.has('s-1'), ).toBe(false); + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started', messageId: 'job-1' }), + expect.objectContaining({ + type: 'cancelled', + messageId: 'job-1', + reason: 'timeout', + }), + ]); } finally { vi.useRealTimers(); } @@ -6973,6 +8503,64 @@ describe('ChannelBase', () => { expect(ch.proactive).toEqual([]); }); + it('does not push a loop response cancelled while waiting for delivery authorization', async () => { + (bridge.prompt as ReturnType).mockResolvedValueOnce( + 'loop response', + ); + let resolveShouldContinue!: (value: boolean) => void; + const shouldContinue = vi + .fn() + .mockResolvedValueOnce(true) + .mockImplementationOnce( + () => + new Promise((resolve) => { + resolveShouldContinue = resolve; + }), + ); + const ch = createChannel(); + ch.enableCancelCommand(); + ch.proactiveSupported = true; + + const loopRun = ch.runLoopPrompt( + { + id: 'job-1', + channelName: 'test-chan', + target: { + channelName: 'test-chan', + senderId: 'alice', + chatId: 'chat1', + isGroup: false, + }, + cwd: '/tmp', + cron: '0 9 * * *', + prompt: 'post summary', + label: 'daily summary', + recurring: true, + enabled: true, + createdBy: 'Alice', + createdAt: '2026-06-30T01:00:00.000Z', + consecutiveFailures: 0, + runCount: 0, + }, + { shouldContinue }, + ); + await vi.waitFor(() => expect(shouldContinue).toHaveBeenCalledTimes(2)); + + await ch.handleInbound(envelope({ text: '/cancel', senderId: 'alice' })); + resolveShouldContinue(true); + + await expect(loopRun).rejects.toThrow('loop cancelled before delivery'); + expect(ch.proactive).toEqual([]); + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started', messageId: 'job-1' }), + expect.objectContaining({ + type: 'cancelled', + messageId: 'job-1', + reason: 'cancel_command', + }), + ]); + }); + it('fails the loop when proactive delivery fails', async () => { (bridge.prompt as ReturnType).mockResolvedValueOnce( 'loop response', @@ -7011,6 +8599,265 @@ describe('ChannelBase', () => { runCount: 0, }), ).rejects.toThrow('api down'); + + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started', messageId: 'job-1' }), + expect.objectContaining({ + type: 'failed', + error: 'api down', + phase: 'delivery', + }), + ]); + }); + + it('emits delivery failure even when a pending cancel settles during wind-down', async () => { + let resolveCancelRpc!: () => void; + (bridge.cancelSession as ReturnType).mockReturnValue( + new Promise((resolve) => { + resolveCancelRpc = resolve; + }), + ); + let releaseShouldContinue!: () => void; + const gate = new Promise((resolve) => { + releaseShouldContinue = resolve; + }); + // First call is the pre-run guard; the second (post-prompt) parks so + // /cancel can land between settle and deliveryStarted. + let shouldContinueCalls = 0; + const shouldContinue = vi.fn().mockImplementation(async () => { + shouldContinueCalls += 1; + if (shouldContinueCalls >= 2) { + await gate; + } + return true; + }); + const ch = createChannel(); + ch.enableCancelCommand(); + ch.proactiveSupported = true; + vi.spyOn( + ch as unknown as { + pushProactive: ( + target: { chatId: string }, + text: string, + ) => Promise; + }, + 'pushProactive', + ).mockRejectedValue(new Error('send failed')); + + const loopRun = ch.runLoopPrompt( + { + id: 'job-1', + channelName: 'test-chan', + target: { + channelName: 'test-chan', + senderId: 'alice', + chatId: 'chat1', + isGroup: false, + }, + cwd: '/tmp', + cron: '0 9 * * *', + prompt: 'post summary', + label: 'daily summary', + recurring: true, + enabled: true, + createdBy: 'Alice', + createdAt: '2026-06-30T01:00:00.000Z', + consecutiveFailures: 0, + runCount: 0, + }, + { shouldContinue }, + ); + loopRun.catch(() => {}); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + + const cancel = ch.handleInbound( + envelope({ text: '/cancel', senderId: 'alice' }), + ); + await vi.waitFor(() => + expect(bridge.cancelSession).toHaveBeenCalledOnce(), + ); + releaseShouldContinue(); + + await expect(loopRun).rejects.toThrow('send failed'); + resolveCancelRpc(); + await cancel; + + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started', messageId: 'job-1' }), + expect.objectContaining({ + type: 'failed', + error: 'send failed', + phase: 'delivery', + }), + ]); + expect(ch.sent).toContainEqual({ + chatId: 'chat1', + text: 'Failed to cancel current request.', + }); + }); + + it('emits failed lifecycle event when a loop prompt rejects', async () => { + (bridge.prompt as ReturnType).mockRejectedValue( + new Error('loop boom'), + ); + const ch = createChannel(); + ch.proactiveSupported = true; + + await expect( + ch.runLoopPrompt({ + id: 'job-1', + channelName: 'test-chan', + target: { + channelName: 'test-chan', + senderId: 'alice', + chatId: 'chat1', + isGroup: false, + }, + cwd: '/tmp', + cron: '0 9 * * *', + prompt: 'post summary', + label: 'daily summary', + recurring: true, + enabled: true, + createdBy: 'Alice', + createdAt: '2026-06-30T01:00:00.000Z', + consecutiveFailures: 0, + runCount: 0, + }), + ).rejects.toThrow('loop boom'); + + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started', messageId: 'job-1' }), + expect.objectContaining({ + type: 'failed', + error: 'loop boom', + phase: 'agent', + messageId: 'job-1', + }), + ]); + }); + + it('emits tool_call lifecycle events during loop prompts', async () => { + let resolvePrompt!: (value: string) => void; + (bridge.prompt as ReturnType).mockReturnValue( + new Promise((resolve) => { + resolvePrompt = resolve; + }), + ); + const ch = createChannel(); + ch.proactiveSupported = true; + + const loopRun = ch.runLoopPrompt({ + id: 'job-1', + channelName: 'test-chan', + target: { + channelName: 'test-chan', + senderId: 'alice', + chatId: 'chat1', + isGroup: false, + }, + cwd: '/tmp', + cron: '0 9 * * *', + prompt: 'post summary', + label: 'daily summary', + recurring: true, + enabled: true, + createdBy: 'Alice', + createdAt: '2026-06-30T01:00:00.000Z', + consecutiveFailures: 0, + runCount: 0, + }); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + + ch.dispatchToolCall({ + sessionId: 's-1', + toolCallId: 'tool-loop', + kind: 'read_file', + title: 'Read README.md', + status: 'running', + rawInput: { path: 'README.md' }, + }); + + const lifecycleToolCall = ch.taskEvents.find( + (event) => event.type === 'tool_call', + ); + expect(lifecycleToolCall).toEqual( + expect.objectContaining({ + type: 'tool_call', + messageId: 'job-1', + toolCall: expect.objectContaining({ toolCallId: 'tool-loop' }), + }), + ); + expect(lifecycleToolCall!.toolCall).not.toHaveProperty('rawInput'); + + resolvePrompt('loop response'); + await loopRun; + }); + + it('completes a loop when cancellation settles after proactive delivery', async () => { + (bridge.prompt as ReturnType).mockResolvedValueOnce( + 'loop response', + ); + let resolveDelivery!: () => void; + const delivery = new Promise((resolve) => { + resolveDelivery = resolve; + }); + const ch = createChannel(); + ch.enableCancelCommand(); + ch.proactiveSupported = true; + vi.spyOn( + ch as unknown as { + pushProactive: ( + target: { chatId: string }, + text: string, + ) => Promise; + }, + 'pushProactive', + ).mockImplementation(async () => { + await delivery; + ch.proactive.push({ chatId: 'chat1', text: 'loop response' }); + }); + + const loopRun = ch.runLoopPrompt({ + id: 'job-1', + channelName: 'test-chan', + target: { + channelName: 'test-chan', + senderId: 'alice', + chatId: 'chat1', + isGroup: false, + }, + cwd: '/tmp', + cron: '0 9 * * *', + prompt: 'post summary', + label: 'daily summary', + recurring: false, + enabled: true, + createdBy: 'Alice', + createdAt: '2026-06-30T01:00:00.000Z', + consecutiveFailures: 0, + runCount: 0, + }); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledOnce()); + await vi.waitFor(() => + expect(bridge.cancelSession).not.toHaveBeenCalled(), + ); + + const cancel = ch.handleInbound( + envelope({ text: '/cancel', senderId: 'alice' }), + ); + await Promise.resolve(); + resolveDelivery(); + await cancel; + + await expect(loopRun).resolves.toBe('loop response'); + expect(ch.proactive).toEqual([ + { chatId: 'chat1', text: 'loop response' }, + ]); + expect(ch.taskEvents).toEqual([ + expect.objectContaining({ type: 'started', messageId: 'job-1' }), + expect.objectContaining({ type: 'completed', messageId: 'job-1' }), + ]); }); it('disables a stored job when its sender is no longer allowed', async () => { diff --git a/packages/channels/base/src/ChannelBase.ts b/packages/channels/base/src/ChannelBase.ts index 473f28c089..469aacb073 100644 --- a/packages/channels/base/src/ChannelBase.ts +++ b/packages/channels/base/src/ChannelBase.ts @@ -3,8 +3,14 @@ import type { ChannelConfig, ChannelMemoryCallbacks, ChannelMemoryTarget, + ChannelRuntimeIdentity, + ChannelRuntimeMemoryScope, + ChannelTaskCancellationReason, + ChannelTaskLifecycleBase, + ChannelTaskLifecycleEvent, DispatchMode, Envelope, + SanitizedToolCallEvent, SessionTarget, } from './types.js'; import { BlockStreamer } from './BlockStreamer.js'; @@ -46,6 +52,8 @@ const CURRENT_MESSAGE_MARKER = '[Current message - respond to this]'; const GROUP_HISTORY_ENTRY_TEXT_LIMIT = 1000; const GROUP_HISTORY_ENTRY_METADATA_LIMIT = 256; const LOOP_CANCEL_GRACE_MS = 5000; +/** Sentinel message for the loop-prompt timeout rejection; matched by identity below. */ +const LOOP_TIMED_OUT_MESSAGE = 'loop timed out'; export interface ChannelBaseOptions { router?: SessionRouter; @@ -84,7 +92,14 @@ export interface ChannelLoopPromptOptions { type CommandHandler = (envelope: Envelope, args: string) => Promise; type ActivePrompt = { cancelled: boolean; + cancelPending?: boolean; + cancellationEmitted?: boolean; cancelRequested?: Promise; + /** Set once response delivery to the platform has begun; past this point a cancel can no longer suppress the turn's output. */ + deliveryStarted?: boolean; + /** Set for loop prompts, whose messageId is an internal job id — adapter + * hooks must not receive it (their contract is platform message ids). */ + loopPrompt?: boolean; done: Promise; resolve: () => void; stopStreaming?: () => void; @@ -150,6 +165,9 @@ export abstract class ChannelBase { protected gate: SenderGate; protected router: SessionRouter; protected name: string; + /** Resolved (defaulted + frozen) identity/scope — adapters should read these, not raw config. */ + protected readonly identity: ChannelRuntimeIdentity; + protected readonly memoryScope: ChannelRuntimeMemoryScope; /** Resolved proxy URL, available to subclasses for adapter-specific clients. */ protected proxy?: string; private readonly channelMemory?: ChannelMemoryCallbacks; @@ -175,10 +193,7 @@ export abstract class ChannelBase { Array<{ text: string; envelope: Envelope }> > = new Map(); private readonly bridgeToolCallListener = (event: ToolCallEvent): void => { - const target = this.router.getTarget(event.sessionId); - if (target) { - this.onToolCall(target.chatId, event); - } + this.dispatchToolCall(event); }; private readonly bridgeSessionDiedListener = ( event: SessionDiedEvent, @@ -186,6 +201,32 @@ export abstract class ChannelBase { this.onSessionDied(event.sessionId); }; + dispatchToolCall(event: ToolCallEvent): void { + const target = this.router.getTarget(event.sessionId); + const active = this.activePrompts.get(event.sessionId); + const chatId = active?.chatId ?? target?.chatId; + if (!chatId) { + return; + } + if (active && !active.cancelled && !active.cancelPending) { + // `?? ''`: dispatchToolCall is a public entry point — a third-party bridge + // omitting a field must not throw out of its emit('toolCall'). + const safeToolCall: SanitizedToolCallEvent = { + sessionId: event.sessionId, + toolCallId: event.toolCallId, + kind: sanitizeLogText(event.kind ?? '', 20), + title: sanitizeLogText(event.title ?? '', 80), + status: sanitizeLogText(event.status ?? '', 20), + }; + this.emitTaskLifecycle({ + ...this.lifecycleBase(chatId, event.sessionId, active.messageId), + type: 'tool_call', + toolCall: safeToolCall, + }); + } + this.onToolCall(chatId, event); + } + constructor( name: string, config: ChannelConfig, @@ -196,6 +237,8 @@ export abstract class ChannelBase { this.config = config; this.bridge = bridge; this.proxy = options?.proxy; + this.identity = Object.freeze(this.resolveIdentity(name, config)); + this.memoryScope = Object.freeze(this.resolveMemoryScope(name, config)); this.channelMemory = options?.channelMemory; this.groupHistory = new GroupHistoryStore( options?.groupHistoryPath ?? @@ -235,6 +278,134 @@ export abstract class ChannelBase { abstract sendMessage(chatId: string, text: string): Promise; abstract disconnect(): void; + /** + * Adapter hook for task lifecycle events. The prompt flow never awaits this + * hook; an async override's rejection is caught and logged, nothing more. + */ + protected onTaskLifecycle( + _event: ChannelTaskLifecycleEvent, + ): void | Promise {} + + private emitTaskLifecycle(event: ChannelTaskLifecycleEvent): void { + try { + const result = this.onTaskLifecycle(event); + if (result && typeof result.catch === 'function') { + result.catch((err: unknown) => { + this.logTaskLifecycleError(event, err); + }); + } + } catch (err) { + this.logTaskLifecycleError(event, err); + } + } + + private logTaskLifecycleError( + event: ChannelTaskLifecycleEvent, + err: unknown, + ): void { + const channel = sanitizeLogText(this.name, 64); + const sessionId = sanitizeLogText(event.sessionId, 64); + const stack = + err instanceof Error && err.stack + ? ` | ${sanitizeLogText(err.stack, 500)}` + : ''; + process.stderr.write( + `[${channel}] onTaskLifecycle threw for ${event.type} session ${sessionId}: ${this.lifecycleError(err)}${stack}\n`, + ); + } + + private lifecycleError(err: unknown): string { + return sanitizeLogText( + err instanceof Error ? err.message : String(err), + 200, + ); + } + + private emitTaskCancellation( + active: ActivePrompt, + sessionId: string, + reason: ChannelTaskCancellationReason, + ): void { + if (active.cancellationEmitted) { + return; + } + active.cancellationEmitted = true; + this.emitTaskLifecycle({ + ...this.lifecycleBase(active.chatId, sessionId, active.messageId), + type: 'cancelled', + reason, + }); + } + + private resolveIdentity( + name: string, + config: ChannelConfig, + ): ChannelRuntimeIdentity { + return { + id: config.identity?.id || `channel:${name}`, + displayName: config.identity?.displayName || name, + ...(config.identity?.description + ? { description: config.identity.description } + : {}), + }; + } + + private resolveMemoryScope( + name: string, + config: ChannelConfig, + ): ChannelRuntimeMemoryScope { + return { + namespace: config.memoryScope?.namespace || `channel:${name}`, + mode: config.memoryScope?.mode ?? 'metadata-only', + }; + } + + /** Built once — identity/memoryScope are frozen at construction. */ + private boundaryPrompt?: string; + + private channelBoundaryPrompt(): string { + if (this.boundaryPrompt !== undefined) { + return this.boundaryPrompt; + } + const identityLines = [ + 'Channel identity:', + `- id: ${sanitizeQuotedText(this.identity.id, 128)}`, + `- display name: ${sanitizeQuotedText(this.identity.displayName, 128)}`, + ...(this.identity.description + ? [ + `- description: ${sanitizeQuotedText(this.identity.description, 256)}`, + ] + : []), + ]; + const memoryLines = [ + 'Memory scope:', + `- namespace: ${sanitizeQuotedText(this.memoryScope.namespace, 128)}`, + `- mode: ${this.memoryScope.mode}`, + '- data from other channels must not be shared.', + ]; + this.boundaryPrompt = [...identityLines, '', ...memoryLines].join('\n'); + return this.boundaryPrompt; + } + + private shouldPrependChannelBoundaryPrompt(): boolean { + return Boolean(this.config.identity || this.config.memoryScope); + } + + private lifecycleBase( + chatId: string, + sessionId: string, + messageId?: string, + ): ChannelTaskLifecycleBase { + return { + channelName: this.name, + chatId, + sessionId, + ...(messageId ? { messageId } : {}), + identity: this.identity, + memoryScope: this.memoryScope, + }; + } + supportsProactiveSend(): boolean { return false; } @@ -359,6 +530,11 @@ export abstract class ChannelBase { if (this.config.instructions) { context.push(this.config.instructions); } + // Boundary block goes last: recency bias means later instructions win, + // and the isolation boundary must not be overridable by operator text. + if (this.shouldPrependChannelBoundaryPrompt()) { + context.push(this.channelBoundaryPrompt()); + } if (context.length > 0) { promptText = `${context.join('\n\n')}\n\n${promptText}`; } @@ -388,13 +564,46 @@ export abstract class ChannelBase { resolve: doneResolve, chatId: job.target.chatId, messageId: job.id, + loopPrompt: true, }; this.activePrompts.set(sessionId, promptState); - this.onPromptStart(job.target.chatId, sessionId); + this.emitTaskLifecycle({ + ...this.lifecycleBase(job.target.chatId, sessionId, job.id), + type: 'started', + }); + // Guarded: an adapter indicator failure must not orphan the started + // event (no terminal) or leak the activePrompts entry. + // No messageId: the hook contract passes INBOUND platform message ids, + // and adapters act on them (cards, reactions) — a loop job id would + // collide. Lifecycle events still carry job.id for correlation. + try { + this.onPromptStart(job.target.chatId, sessionId); + } catch (err) { + process.stderr.write( + `[${this.name}] onPromptStart threw in loop ${job.id} for session ${sessionId}: ${this.lifecycleError(err)}\n`, + ); + } + // Same hold-and-replay contract as handleInbound's onChunk: visible + // sinks stay out of the transcript while a cancel is pending. + const heldChunks: string[] = []; + const releaseHeldChunks = () => { + for (const held of heldChunks.splice(0)) { + this.emitTaskLifecycle({ + ...this.lifecycleBase(job.target.chatId, sessionId, job.id), + type: 'text_chunk', + chunk: held, + }); + this.onResponseChunk(job.target.chatId, held, sessionId); + } + }; const onChunk = (sid: string, chunk: string) => { - if (sid === sessionId && !promptState.cancelled) { - this.onResponseChunk(job.target.chatId, chunk, sessionId); + if (sid !== sessionId || promptState.cancelled) { + return; + } + heldChunks.push(chunk); + if (!promptState.cancelPending) { + releaseHeldChunks(); } }; const promptBridge = this.bridge; @@ -409,22 +618,83 @@ export abstract class ChannelBase { job.id, options.timeoutMs, ); - if (promptState.cancelRequested && !promptState.cancelled) { - const cancelled = await promptState.cancelRequested; - if (cancelled) { - promptState.cancelled = true; - } - } + await this.settleCancelRequested(promptState); if (promptState.cancelled) { - throw new ChannelLoopSkippedError('loop cancelled before delivery'); + throw new ChannelLoopSkippedError( + 'loop cancelled before delivery', + 'cancel_command', + ); } + releaseHeldChunks(); if (options.shouldContinue && !(await options.shouldContinue())) { throw new ChannelLoopSkippedError('loop dropped before delivery'); } + if (promptState.cancelled) { + throw new ChannelLoopSkippedError( + 'loop cancelled before delivery', + 'cancel_command', + ); + } if (response) { + promptState.deliveryStarted = true; await this.pushProactive(job.target, response); } + // Once delivery started the run counts as completed — a cancel settling + // during/after the send must not convert a delivered run into a skip + // (a one-shot loop would stay enabled and deliver twice). + if (!promptState.deliveryStarted) { + await this.settleCancelRequested(promptState); + if (promptState.cancelled) { + throw new ChannelLoopSkippedError( + 'loop cancelled before delivery', + 'cancel_command', + ); + } + } + // /clear can evict mid-delivery and emit its own terminal event; never + // follow a cancelled event with completed for the same prompt. + if (!promptState.cancellationEmitted) { + this.emitTaskLifecycle({ + ...this.lifecycleBase(job.target.chatId, sessionId, job.id), + type: 'completed', + }); + } return response; + } catch (err) { + // Once delivery started, a late-settling cancel must not flip + // `cancelled` here — it would suppress the failed emit while the + // /cancel handler (seeing deliveryStarted) declines to emit its own + // terminal, leaving the task with no terminal event at all. + if (!promptState.deliveryStarted) { + await this.settleCancelRequested(promptState); + } + if (err instanceof ChannelLoopSkippedError && !promptState.cancelled) { + this.emitTaskCancellation(promptState, sessionId, err.reason); + promptState.cancelled = true; + } + if ( + !promptState.cancelled && + !(err instanceof ChannelLoopSkippedError) + ) { + this.emitTaskLifecycle({ + ...this.lifecycleBase(job.target.chatId, sessionId, job.id), + type: 'failed', + error: this.lifecycleError(err), + phase: promptState.deliveryStarted ? 'delivery' : 'agent', + }); + } else if ( + promptState.cancelled && + !(err instanceof ChannelLoopSkippedError) && + !(err instanceof Error && err.message === LOOP_TIMED_OUT_MESSAGE) + ) { + const channel = sanitizeLogText(this.name, 64); + const safeJobId = sanitizeLogText(job.id, 64); + const safeSessionId = sanitizeLogText(sessionId, 64); + process.stderr.write( + `[${channel}] loop ${safeJobId} threw after cancellation for session ${safeSessionId}: ${this.lifecycleError(err)}\n`, + ); + } + throw err; } finally { promptBridge.off('textChunk', onChunk); const stillCurrent = this.activePrompts.get(sessionId) === promptState; @@ -493,15 +763,16 @@ export abstract class ChannelBase { prompt, new Promise((_, reject) => { timer = setTimeout(() => { - reject(new Error('loop timed out')); + reject(new Error(LOOP_TIMED_OUT_MESSAGE)); }, timeoutMs); timer.unref?.(); }), ]); } catch (err) { - if (err instanceof Error && err.message === 'loop timed out') { + if (err instanceof Error && err.message === LOOP_TIMED_OUT_MESSAGE) { promptState.cancelled = true; await this.cancelTimedOutLoopPrompt(promptBridge, sessionId, jobId); + this.emitTaskCancellation(promptState, sessionId, 'timeout'); } throw err; } finally { @@ -541,6 +812,27 @@ export abstract class ChannelBase { } } + private async settleCancelRequested(active: ActivePrompt): Promise { + if (!active.cancelRequested || active.cancelled) { + return; + } + let timer: ReturnType | undefined; + try { + const cancelled = await Promise.race([ + active.cancelRequested, + new Promise((resolve) => { + timer = setTimeout(() => resolve(false), CLEAR_CANCEL_TIMEOUT_MS); + timer.unref?.(); + }), + ]); + if (cancelled) { + active.cancelled = true; + } + } finally { + clearTimeout(timer); + } + } + onToolCall(_chatId: string, _event: ToolCallEvent): void {} onSessionDied(sessionId: string): void { @@ -642,6 +934,13 @@ export abstract class ChannelBase { ); return true; } + if (active.deliveryStarted) { + await this.sendMessage( + envelope.chatId, + 'Failed to cancel current request.', + ); + return true; + } const cancelRequested = active.cancelRequested ?? @@ -649,16 +948,29 @@ export abstract class ChannelBase { () => true, (err) => { process.stderr.write( - `[${this.name}] cancelSession failed for session=${activeSessionId}: ${err instanceof Error ? err.message : err}\n`, + `[${sanitizeLogText(this.name, 64)}] cancelSession failed for session=${sanitizeLogText(activeSessionId, 64)}: ${this.lifecycleError(err)}\n`, ); active.cancelRequested = undefined; return false; }, ); active.cancelRequested = cancelRequested; + active.cancelPending = true; - const cancelSucceeded = await cancelRequested; - if (!cancelSucceeded) { + let cancelSucceeded: boolean; + try { + cancelSucceeded = await cancelRequested; + } finally { + active.cancelPending = false; + } + // Re-check after the await: the turn may have ended (or its delivery + // begun) while the cancel RPC was in flight — claiming success then + // would emit a spurious cancelled event for a delivered response. + if ( + this.activePrompts.get(activeSessionId) !== active || + active.deliveryStarted || + !cancelSucceeded + ) { await this.sendMessage( envelope.chatId, 'Failed to cancel current request.', @@ -667,8 +979,9 @@ export abstract class ChannelBase { } active.cancelled = true; - active.stopStreaming?.(); + this.stopActiveStreaming(active, activeSessionId, 'cancel'); this.collectBuffers.delete(activeSessionId); + this.emitTaskCancellation(active, activeSessionId, 'cancel_command'); await this.sendMessage(envelope.chatId, 'Cancelled current request.'); return true; }); @@ -743,7 +1056,11 @@ export abstract class ChannelBase { // onPromptEnd anyway. Letting the purge proceed makes the turn // non-current, so the clearEvicted guard then skips correctly. try { - this.onPromptEnd(active.chatId, id, active.messageId); + this.onPromptEnd( + active.chatId, + id, + active.loopPrompt ? undefined : active.messageId, + ); } catch (err) { process.stderr.write( `[${this.name}] onPromptEnd threw during /clear eviction for session ${id}: ${err instanceof Error ? err.message : err}\n`, @@ -853,6 +1170,14 @@ export abstract class ChannelBase { envelope.chatId, [ `Channel: ${this.name}`, + // Identity/memory lines only for channels that opted in — keep + // unconfigured channels' output unchanged. + ...(this.shouldPrependChannelBoundaryPrompt() + ? [ + `Identity: ${sanitizeQuotedText(this.identity.displayName, 128)}`, + `Memory: ${sanitizeQuotedText(this.memoryScope.namespace, 128)}`, + ] + : []), // Only the basename — don't leak the absolute cwd to group members. `Workspace: ${basename(this.config.cwd)}`, `Session: ${active ? 'active' : 'none'}${scopeNote}`, @@ -1061,6 +1386,12 @@ export abstract class ChannelBase { `Session: ${hasSession ? 'active' : 'none'}`, `Access: ${policy}`, `Channel: ${this.name}`, + ...(this.shouldPrependChannelBoundaryPrompt() + ? [ + `Identity: ${sanitizeQuotedText(this.identity.id, 128)}`, + `Memory: ${this.memoryScope.mode}`, + ] + : []), ]; await this.sendMessage(envelope.chatId, lines.join('\n')); return true; @@ -1540,6 +1871,7 @@ export abstract class ChannelBase { `[${this.name}] cancelSession failed for session=${sessionId} (clear/await): ${err instanceof Error ? err.message : err}\n`, ); }); + this.emitTaskCancellation(active, sessionId, 'clear'); let timer: ReturnType | undefined; const settled = await Promise.race([ active.done.then(() => true), @@ -2052,18 +2384,25 @@ export abstract class ChannelBase { // turn that runs without waiting for a wedged predecessor) is the // deferred fix — it needs an API change across every adapter and is out // of scope for this phase (wenshao option (b)). + const firstCancellation = !active.cancelled; active.cancelled = true; - process.stderr.write( - `[${this.name}] steer: cancelled active turn for ${envelope.senderId} in session ${sessionId}\n`, - ); - this.stopActiveStreaming(active, sessionId, 'steer'); - // Fire-and-forget, but LOG the IPC failure rather than swallow it, so a - // best-effort cancel that fails isn't silently invisible to operators. - void this.bridge.cancelSession(sessionId).catch((err) => { + if (firstCancellation) { process.stderr.write( - `[${this.name}] cancelSession failed for session=${sessionId} (steer): ${err instanceof Error ? err.message : err}\n`, + `[${this.name}] steer: cancelled active turn for ${envelope.senderId} in session ${sessionId}\n`, ); - }); + this.stopActiveStreaming(active, sessionId, 'steer'); + // Fire-and-forget, but LOG the IPC failure rather than swallow it, so a + // best-effort cancel that fails isn't silently invisible to operators. + void this.bridge.cancelSession(sessionId).catch((err) => { + process.stderr.write( + `[${this.name}] cancelSession failed for session=${sessionId} (steer): ${err instanceof Error ? err.message : err}\n`, + ); + }); + // Emitted before the bridge cancel settles: steer supersedes the + // turn at the channel level (cancelled is already set above), so + // the event reflects that intent, not the bridge RPC outcome. + this.emitTaskCancellation(active, sessionId, 'steer'); + } // Diagnostic watchdog: if the predecessor turn is STILL the active prompt // after the wind-down bound, this steered turn is wedged behind a hung // bridge.prompt() — surface it (the chained `.then()` clears it once the @@ -2161,6 +2500,11 @@ export abstract class ChannelBase { if (this.config.instructions) { sessionContext.push(this.config.instructions); } + // Boundary block goes last: recency bias means later instructions win, + // and the isolation boundary must not be overridable by operator text. + if (this.shouldPrependChannelBoundaryPrompt()) { + sessionContext.push(this.channelBoundaryPrompt()); + } } if (this.dropQueuedTurnIfStale(sessionId, generation, envelope)) { return; @@ -2191,8 +2535,20 @@ export abstract class ChannelBase { // (Steer no longer hands a still-active session to a replacement; only // /clear evicts, and it gives the next turn a fresh session.) this.activePrompts.set(sessionId, promptState); + this.emitTaskLifecycle({ + ...this.lifecycleBase(envelope.chatId, sessionId, envelope.messageId), + type: 'started', + }); - this.onPromptStart(envelope.chatId, sessionId, envelope.messageId); + // Guarded: an adapter indicator failure must not orphan the started + // event (no terminal) or leak the activePrompts entry. + try { + this.onPromptStart(envelope.chatId, sessionId, envelope.messageId); + } catch (err) { + process.stderr.write( + `[${this.name}] onPromptStart threw for session ${sessionId}: ${this.lifecycleError(err)}\n`, + ); + } const streamer = useBlockStreaming ? new BlockStreamer({ @@ -2204,10 +2560,32 @@ export abstract class ChannelBase { : null; promptState.stopStreaming = () => streamer?.stop(); + // Chunks arriving while a cancel is PENDING are held here: pushing them + // to any visible sink could send output the cancel can't recall. On a + // failed cancel they're replayed; on success, discarded. + const heldChunks: string[] = []; + const releaseHeldChunks = () => { + for (const held of heldChunks.splice(0)) { + this.emitTaskLifecycle({ + ...this.lifecycleBase( + envelope.chatId, + sessionId, + envelope.messageId, + ), + type: 'text_chunk', + chunk: held, + }); + this.onResponseChunk(envelope.chatId, held, sessionId); + streamer?.push(held); + } + }; const onChunk = (sid: string, chunk: string) => { - if (sid === sessionId && !promptState.cancelled) { - this.onResponseChunk(envelope.chatId, chunk, sessionId); - streamer?.push(chunk); + if (sid !== sessionId || promptState.cancelled) { + return; + } + heldChunks.push(chunk); + if (!promptState.cancelPending) { + releaseHeldChunks(); } }; const promptBridge = this.bridge; @@ -2219,21 +2597,62 @@ export abstract class ChannelBase { imageMimeType, }); - if (promptState.cancelRequested && !promptState.cancelled) { - const cancelled = await promptState.cancelRequested; - if (cancelled) { - promptState.cancelled = true; - } + await this.settleCancelRequested(promptState); + if (!promptState.cancelled) { + releaseHeldChunks(); } // If cancelled, skip sending the response if (!promptState.cancelled && response) { + promptState.deliveryStarted = true; if (streamer) { await streamer.flush(); } else { await this.onResponseComplete(envelope.chatId, response, sessionId); } } + // Once delivery started the turn's outcome is fixed — don't let a + // cancel settling during the send rewrite completed into cancelled. + if (!promptState.deliveryStarted) { + await this.settleCancelRequested(promptState); + } + if (!promptState.cancelled && !promptState.cancellationEmitted) { + this.emitTaskLifecycle({ + ...this.lifecycleBase( + envelope.chatId, + sessionId, + envelope.messageId, + ), + type: 'completed', + }); + } + } catch (err) { + // Mirror the try path: once delivery started, a late-settling cancel + // must not suppress the failed emit (the /cancel handler declines to + // emit its own terminal once deliveryStarted is set). + if (!promptState.deliveryStarted) { + await this.settleCancelRequested(promptState); + } + if (!promptState.cancelled) { + this.emitTaskLifecycle({ + ...this.lifecycleBase( + envelope.chatId, + sessionId, + envelope.messageId, + ), + type: 'failed', + error: this.lifecycleError(err), + phase: promptState.deliveryStarted ? 'delivery' : 'agent', + }); + } else { + const channel = sanitizeLogText(this.name, 64); + const safeSessionId = sanitizeLogText(sessionId, 64); + const safeMessageId = sanitizeLogText(envelope.messageId ?? '', 64); + process.stderr.write( + `[${channel}] turn ${safeMessageId} threw after cancellation for session ${safeSessionId}: ${this.lifecycleError(err)}\n`, + ); + } + throw err; } finally { promptBridge.off('textChunk', onChunk); streamer?.stop(); diff --git a/packages/channels/base/src/ChannelLoopScheduler.ts b/packages/channels/base/src/ChannelLoopScheduler.ts index 94f1a312d7..091c819e3b 100644 --- a/packages/channels/base/src/ChannelLoopScheduler.ts +++ b/packages/channels/base/src/ChannelLoopScheduler.ts @@ -24,7 +24,17 @@ export interface ChannelLoopSchedulerOptions { loopTimeoutMs?: number; } -export class ChannelLoopSkippedError extends Error {} +/** Why a loop run was skipped; carried as data so reporting never depends on message wording. */ +export type ChannelLoopSkipReason = 'cancel_command' | 'clear' | 'dropped'; + +export class ChannelLoopSkippedError extends Error { + constructor( + message: string, + readonly reason: ChannelLoopSkipReason = 'dropped', + ) { + super(message); + } +} export class ChannelLoopScheduler { private readonly store: Pick; diff --git a/packages/channels/base/src/index.ts b/packages/channels/base/src/index.ts index a997edb631..1cefb4a420 100644 --- a/packages/channels/base/src/index.ts +++ b/packages/channels/base/src/index.ts @@ -55,12 +55,21 @@ export type { BlockStreamingChunkConfig, BlockStreamingCoalesceConfig, ChannelConfig, + ChannelIdentityConfig, + ChannelMemoryScopeConfig, + ChannelMemoryScopeMode, ChannelPlugin, + ChannelRuntimeIdentity, + ChannelRuntimeMemoryScope, + ChannelTaskCancellationReason, + ChannelTaskLifecycleBase, + ChannelTaskLifecycleEvent, ChannelType, DispatchMode, Envelope, GroupConfig, GroupPolicy, + SanitizedToolCallEvent, SenderPolicy, SessionScope, SessionTarget, diff --git a/packages/channels/base/src/types.ts b/packages/channels/base/src/types.ts index 8830107ea5..6612274c15 100644 --- a/packages/channels/base/src/types.ts +++ b/packages/channels/base/src/types.ts @@ -7,6 +7,30 @@ export type ChannelType = string; export type GroupPolicy = 'disabled' | 'allowlist' | 'open'; export type DispatchMode = 'collect' | 'steer' | 'followup'; +export interface ChannelIdentityConfig { + id?: string; + displayName?: string; + description?: string; +} + +export interface ChannelRuntimeIdentity { + readonly id: string; + readonly displayName: string; + readonly description?: string; +} + +export type ChannelMemoryScopeMode = 'metadata-only'; + +export interface ChannelMemoryScopeConfig { + namespace?: string; + mode?: ChannelMemoryScopeMode; +} + +export interface ChannelRuntimeMemoryScope { + readonly namespace: string; + readonly mode: ChannelMemoryScopeMode; +} + export interface GroupConfig { requireMention?: boolean; // default: true dispatchMode?: DispatchMode; @@ -36,6 +60,8 @@ export interface ChannelConfig { cwd: string; approvalMode?: string; instructions?: string; + identity?: ChannelIdentityConfig; + memoryScope?: ChannelMemoryScopeConfig; model?: string; groupPolicy: GroupPolicy; // default: "disabled" groupHistoryLimit?: number; @@ -105,6 +131,55 @@ export interface SessionTarget { isGroup?: boolean; } +export interface ChannelTaskLifecycleBase { + channelName: string; + chatId: string; + sessionId: string; + messageId?: string; + identity: ChannelRuntimeIdentity; + memoryScope: ChannelRuntimeMemoryScope; +} + +/** + * Whitelist of tool-call fields exposed to lifecycle consumers. Kept explicit + * (not derived from ToolCallEvent) so a new bridge field can't leak through. + */ +export interface SanitizedToolCallEvent { + sessionId: string; + toolCallId: string; + kind: string; + title: string; + status: string; +} + +/** 'dropped' = loop was disabled/deleted mid-run (not user-cancelled). */ +export type ChannelTaskCancellationReason = + | 'cancel_command' + | 'clear' + | 'steer' + | 'timeout' + | 'dropped'; + +export type ChannelTaskLifecycleEvent = + | (ChannelTaskLifecycleBase & { type: 'started' }) + /** `chunk` is raw model output — content, not metadata; deliberately unsanitized. */ + | (ChannelTaskLifecycleBase & { type: 'text_chunk'; chunk: string }) + | (ChannelTaskLifecycleBase & { + type: 'tool_call'; + toolCall: SanitizedToolCallEvent; + }) + | (ChannelTaskLifecycleBase & { + type: 'cancelled'; + reason: ChannelTaskCancellationReason; + }) + | (ChannelTaskLifecycleBase & { type: 'completed' }) + | (ChannelTaskLifecycleBase & { + type: 'failed'; + error: string; + /** Where the turn failed: agent generation vs delivery to the platform. */ + phase: 'agent' | 'delivery'; + }); + export interface ChannelMemoryTarget { channelName: string; chatId: string; diff --git a/packages/cli/src/commands/channel/config-utils.test.ts b/packages/cli/src/commands/channel/config-utils.test.ts index 4ac5481fff..98e4063e6c 100644 --- a/packages/cli/src/commands/channel/config-utils.test.ts +++ b/packages/cli/src/commands/channel/config-utils.test.ts @@ -105,6 +105,8 @@ describe('parseChannelConfig', () => { expect(result.cwd).toBe(process.cwd()); expect(result.groupPolicy).toBe('disabled'); expect(result.groups).toEqual({}); + expect(result.identity).toBeUndefined(); + expect(result.memoryScope).toBeUndefined(); }); it('resolves env vars in token, clientId, clientSecret', async () => { @@ -138,6 +140,8 @@ describe('parseChannelConfig', () => { cwd: '/custom', approvalMode: 'auto', instructions: 'Be helpful', + identity: { id: 'ops-agent', displayName: 'Ops Agent' }, + memoryScope: { namespace: 'qwen-tag:ops', mode: 'metadata-only' }, model: 'qwen-coder', groupPolicy: 'open', groups: { g1: { mentionKeywords: ['@bot'] } }, @@ -150,11 +154,56 @@ describe('parseChannelConfig', () => { expect(result.cwd).toBe('/custom'); expect(result.approvalMode).toBe('auto'); expect(result.instructions).toBe('Be helpful'); + expect(result.identity).toEqual({ + id: 'ops-agent', + displayName: 'Ops Agent', + }); + expect(result.memoryScope).toEqual({ + namespace: 'qwen-tag:ops', + mode: 'metadata-only', + }); expect(result.model).toBe('qwen-coder'); expect(result.groupPolicy).toBe('open'); expect(result.groups).toEqual({ g1: { mentionKeywords: ['@bot'] } }); }); + it('rejects a non-object identity', async () => { + await expect( + parseChannelConfig('bot', { type: 'bare', identity: 'ops' }), + ).rejects.toThrow('Channel "bot" field "identity" must be an object.'); + }); + + it('rejects a non-string identity field', async () => { + await expect( + parseChannelConfig('bot', { type: 'bare', identity: { id: 123 } }), + ).rejects.toThrow('Channel "bot" field "identity.id" must be a string.'); + }); + + it('rejects a non-object memoryScope', async () => { + await expect( + parseChannelConfig('bot', { type: 'bare', memoryScope: ['ops'] }), + ).rejects.toThrow('Channel "bot" field "memoryScope" must be an object.'); + }); + + it('rejects an unknown memoryScope.mode', async () => { + await expect( + parseChannelConfig('bot', { + type: 'bare', + memoryScope: { mode: 'full' }, + }), + ).rejects.toThrow( + 'Channel "bot" field "memoryScope.mode" must be "metadata-only".', + ); + }); + + it('drops empty identity fields instead of failing', async () => { + const result = await parseChannelConfig('bot', { + type: 'bare', + identity: { id: 'ops-agent', displayName: '', description: null }, + }); + expect(result.identity).toEqual({ id: 'ops-agent' }); + }); + it('spreads extra fields from raw config', async () => { const result = await parseChannelConfig('bot', { type: 'bare', diff --git a/packages/cli/src/commands/channel/config-utils.ts b/packages/cli/src/commands/channel/config-utils.ts index 6f6c6c49b0..b21ae7f0c6 100644 --- a/packages/cli/src/commands/channel/config-utils.ts +++ b/packages/cli/src/commands/channel/config-utils.ts @@ -35,6 +35,61 @@ function resolveOptionalStringField( return resolveEnvVars(value); } +/** + * Validate identity/memoryScope shape at parse time. settings.json is + * hand-edited; a malformed value would otherwise surface as an opaque + * TypeError on the first prompt of every session instead of at startup. + */ +function parseObjectStringFields( + channelName: string, + rawConfig: Record, + key: 'identity' | 'memoryScope', + fields: readonly Field[], +): Record | undefined { + const value = rawConfig[key]; + if (value === undefined || value === null) { + return undefined; + } + if (typeof value !== 'object' || Array.isArray(value)) { + throw new Error( + `Channel "${channelName}" field "${key}" must be an object.`, + ); + } + const record = value as Record; + const result: Record = {}; + for (const field of fields) { + const fieldValue = record[field]; + if (fieldValue === undefined || fieldValue === null || fieldValue === '') { + continue; + } + if (typeof fieldValue !== 'string') { + throw new Error( + `Channel "${channelName}" field "${key}.${field}" must be a string.`, + ); + } + result[field] = fieldValue; + } + return result; +} + +function parseMemoryScopeConfig( + channelName: string, + rawConfig: Record, +): ChannelConfig['memoryScope'] { + const parsed = parseObjectStringFields( + channelName, + rawConfig, + 'memoryScope', + ['namespace', 'mode'] as const, + ); + if (parsed?.['mode'] !== undefined && parsed['mode'] !== 'metadata-only') { + throw new Error( + `Channel "${channelName}" field "memoryScope.mode" must be "metadata-only".`, + ); + } + return parsed as ChannelConfig['memoryScope']; +} + export async function parseChannelConfig( name: string, rawConfig: Record, @@ -87,6 +142,12 @@ export async function parseChannelConfig( cwd: resolvePath((rawConfig['cwd'] as string) || defaultCwd), approvalMode: rawConfig['approvalMode'] as string | undefined, instructions: rawConfig['instructions'] as string | undefined, + identity: parseObjectStringFields(name, rawConfig, 'identity', [ + 'id', + 'displayName', + 'description', + ] as const) as ChannelConfig['identity'], + memoryScope: parseMemoryScopeConfig(name, rawConfig), model: rawConfig['model'] as string | undefined, groupPolicy: (rawConfig['groupPolicy'] as ChannelConfig['groupPolicy']) || 'disabled', diff --git a/packages/cli/src/commands/channel/runtime.ts b/packages/cli/src/commands/channel/runtime.ts index f5a22822ab..3b85c64c82 100644 --- a/packages/cli/src/commands/channel/runtime.ts +++ b/packages/cli/src/commands/channel/runtime.ts @@ -158,7 +158,7 @@ export function registerToolCallDispatch( if (target) { const channel = channels.get(target.channelName); if (channel) { - channel.onToolCall(target.chatId, event); + channel.dispatchToolCall(event); } } }); diff --git a/packages/cli/src/commands/channel/start.test.ts b/packages/cli/src/commands/channel/start.test.ts index 03b99046b9..2384c95624 100644 --- a/packages/cli/src/commands/channel/start.test.ts +++ b/packages/cli/src/commands/channel/start.test.ts @@ -38,6 +38,7 @@ const mockChannelConnect = vi.hoisted(() => vi.fn()); const mockChannelDisconnect = vi.hoisted(() => vi.fn()); const mockChannelSetBridge = vi.hoisted(() => vi.fn()); const mockChannelOnToolCall = vi.hoisted(() => vi.fn()); +const mockChannelDispatchToolCall = vi.hoisted(() => vi.fn()); const mockChannelOnSessionDied = vi.hoisted(() => vi.fn()); const mockCreateChannel = vi.hoisted(() => vi.fn()); const mockBridgeStart = vi.hoisted(() => vi.fn()); @@ -176,6 +177,7 @@ const mockChannel = { disconnect: mockChannelDisconnect, onSessionDied: mockChannelOnSessionDied, onToolCall: mockChannelOnToolCall, + dispatchToolCall: mockChannelDispatchToolCall, setBridge: mockChannelSetBridge, }; @@ -584,7 +586,8 @@ describe('startCommand.handler', () => { toolCallListener!(event); expect(mockRouterGetTarget).toHaveBeenCalledWith('s-1'); - expect(mockChannelOnToolCall).toHaveBeenCalledWith('chat1', event); + expect(mockChannelDispatchToolCall).toHaveBeenCalledWith(event); + expect(mockChannelOnToolCall).not.toHaveBeenCalled(); }); it('dispatches session death to the owning channel when the route is known', async () => {