vrr/qwen-code
2
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-07-09 17:19:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

feat(channels): add identity and task lifecycle metadata (#6105)

Browse source 
* 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 <qwen-coder@alibabacloud.com>

* 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 <qwen-coder@alibabacloud.com>

* 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 <qwen-coder@alibabacloud.com>

* fix(channels): defer adapter chunks while cancel is pending

---------

Co-authored-by: Shaojin Wen <shaojin.wensj@alibaba-inc.com>
Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
This commit is contained in:
qqqys 2026-07-02 17:04:07 +08:00 • committed by GitHub
parent 230f4af276
commit ca61d7827e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 3524 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -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

735
docs/superpowers/plans/2026-07-01-channel-p0-identity-task-lifecycle.md Normal file
View file

@ -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<ChannelTaskLifecycleBase, 'messageId'> & {
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.

272
docs/superpowers/specs/2026-07-01-channel-p0-identity-task-lifecycle-design.md Normal file
View file

@ -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:<name>`
- `displayName`: `config.identity.displayName` or `<name>`
- `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:<name>`
- `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.

1849
packages/channels/base/src/ChannelBase.test.ts
View file

File diff suppressed because it is too large Load diff

497
packages/channels/base/src/ChannelBase.ts
View file

@ -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<boolean>;
type ActivePrompt = {
cancelled: boolean;
cancelPending?: boolean;
cancellationEmitted?: boolean;
cancelRequested?: Promise<boolean>;
/** 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<void>;
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<void>;
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<void> {}
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<never>((_, 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<void> {
if (!active.cancelRequested || active.cancelled) {
return;
}
let timer: ReturnType<typeof setTimeout> | undefined;
try {
const cancelled = await Promise.race([
active.cancelRequested,
new Promise<boolean>((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<typeof setTimeout> | 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();

12
packages/channels/base/src/ChannelLoopScheduler.ts
View file

@ -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<ChannelLoopStore, 'list' | 'update' | 'disable'>;

9
packages/channels/base/src/index.ts
View file

@ -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,

75
packages/channels/base/src/types.ts
View file

@ -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;

49
packages/cli/src/commands/channel/config-utils.test.ts
View file

@ -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',

61
packages/cli/src/commands/channel/config-utils.ts
View file

@ -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<Field extends string>(
channelName: string,
rawConfig: Record<string, unknown>,
key: 'identity' | 'memoryScope',
fields: readonly Field[],
): Record<string, string> | 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<string, unknown>;
const result: Record<string, string> = {};
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<string, unknown>,
): 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<string, unknown>,
@ -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',

2
packages/cli/src/commands/channel/runtime.ts
View file

@ -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);
}
}
});

5
packages/cli/src/commands/channel/start.test.ts
View file

@ -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 () => {