From 3a24dc4d6c21db57784ec26c5687affe5b2b76b5 Mon Sep 17 00:00:00 2001 From: Ying Xiang Date: Thu, 20 Aug 2026 12:55:47 +0800 Subject: [PATCH 1/3] feat(pi-agent-team): separate team and round lifecycle --- packages/pi-agent-team/README.md | 9 +- .../docs/lifecycle-improvement-evaluation.md | 138 +++++++ packages/pi-agent-team/src/extension.ts | 44 ++- packages/pi-agent-team/src/run-manager.ts | 365 +++++++++--------- .../pi-agent-team/test/run-manager.test.ts | 76 +++- 5 files changed, 444 insertions(+), 188 deletions(-) create mode 100644 packages/pi-agent-team/docs/lifecycle-improvement-evaluation.md diff --git a/packages/pi-agent-team/README.md b/packages/pi-agent-team/README.md index 09fe4d5..d3e6589 100644 --- a/packages/pi-agent-team/README.md +++ b/packages/pi-agent-team/README.md @@ -20,6 +20,12 @@ pi -e npm:@geminixiang/pi-agent-team `pi remove npm:@geminixiang/pi-agent-team` uninstalls it. Use `pi install -l` to write to project settings (`.pi/settings.json`) instead of user settings. +## Lifecycle API migration + +Existing callers do not need to change the handle they store: the tool parameter remains named `runId` for compatibility, but its value now identifies the retained team (`teamId`), not an individual execution. Use snapshot `roundId`/`roundIndex` to correlate one objective execution. `status` remains the latest round status; use `lifecycle` to distinguish an `available`, `running`, `closing`, or `closed` retained team. Terminal `rounds` are bounded to the latest 16 summaries, and restricted message bodies are never included. + +`team_cancel` is idempotent for the latest cancelled round. Callers that retry asynchronously may pass the observed optional `roundId`; if a newer round has started, the stale cancellation is a no-op. Once cancellation settles, `team_prompt` starts a fresh round on the same member `sessionId`/`sessionRef` values. Events and late runtime callbacks are round-scoped so an older completion cannot settle or modify a newer round. + ## What ships The bundled `pi-agent-team` skill teaches the parent agent when to use foreground or detached mode, how to observe and intervene without polling or unsolicited guidance, and how to continue a settled retained team. It is withheld from member sessions, which lack `team_start`, while their other installed skills remain available. @@ -107,7 +113,8 @@ These tests prove the generic runtime routes isolated adapters correctly. They d ## Limits - A `completed` settlement means only that all members called `team_finish`; it is not a scenario-specific correctness verdict. A `quiescent` settlement means no members are runnable while unfinished members remain — `errored-members-remain` is terminal failure, `blocked-members-remain` is an explicit requester wait (for non-detached runs), and `no-runnable-members` may be a genuine deadlock. An `exhausted` settlement means the turn budget ran out with runnable members remaining, enforced as a real per-wave hard cap rather than only checked before a wave starts; the partial result is still returned. Inspect public speech and causal evidence before claiming that the objective succeeded. -- A synchronous `team_start` remains foreground until its first round settles. Its manifest's `team` id is then accepted by `team_get` and `team_prompt`; prompting a settled member continues the retained team in a new background round. Retained teams and their controls are process/session-local: no daemon, socket API, or restart persistence. A `TeamRuntime` instance still represents exactly one round; continuation creates a clean runtime over the retained agent sessions rather than reusing settled coordination state. +- **Stable team identity, explicit round identity.** Control tools continue to accept the original `runId` parameter as a stable `teamId` handle. Every objective execution has a unique `roundId` and monotonic `roundIndex`; snapshots expose `lifecycle`, the current/latest round, cancellation metadata, and capped terminal summaries. +- A synchronous `team_start` remains foreground until its first round settles. Its manifest's stable `team` id is then accepted by `team_get` and `team_prompt`; prompting an available team continues the retained team in a new background round. Retained teams and their controls are process/session-local: no daemon, socket API, or restart persistence. A `TeamRuntime` instance still represents exactly one round; continuation creates a clean runtime over the retained agent sessions rather than reusing settled coordination state. Cancellation settles only the current round; the retained member sessions remain available. - A quiescent team may stop before every member calls `team_finish`; this is reported as `quiescent`, never silently called success. - The parent supplies the initial objective/message. Runtime-generated follow-up hints and scenario-specific fallback decisions do not exist; runtime messages are limited to task-agnostic control notices (claim results, bounces, budget and error alerts). - A command batch returned by one turn first passes a hard freshness fence: if an authorized observation arrived for that member during `act()`, none of the stale batch commits and the member rethinks from all new observations. Otherwise it is applied in order and fail-stop, not as a transaction: commands that already committed are never rolled back, but the first rejected command halts the batch and everything after it is discarded. The rejection bounces back (`COMMAND_FAILED`, plus `COMMAND_BATCH_HALTED` when later commands were discarded) and wakes the member to replan next turn — so a rejected send can no longer be sealed off by a `finish` queued behind it in the same batch. A genuine cross-command dependency still needs a cross-turn fence (a claim, or waiting for a confirmation message), not same-batch ordering. diff --git a/packages/pi-agent-team/docs/lifecycle-improvement-evaluation.md b/packages/pi-agent-team/docs/lifecycle-improvement-evaluation.md new file mode 100644 index 0000000..3748474 --- /dev/null +++ b/packages/pi-agent-team/docs/lifecycle-improvement-evaluation.md @@ -0,0 +1,138 @@ +# Agent Team Lifecycle Improvement Evaluation + +## Decision under evaluation + +Separate a retained team identity from each objective execution: + +- `TeamHandle`: stable roster and retained member sessions. +- `RoundRun`: one objective, one execution lifecycle, and one outcome. +- Cancelling a round must not destroy the retained team; a later prompt can start a new round. + +This evaluation is intentionally domain-neutral. Werewolf is one stress scenario, not part of the production model. + +## Why this is first + +Today one run identifier and one status surface conflate the retained team, its current round, and the latest outcome. That makes cancellation and continuation ambiguous. The change is successful only if it makes those states explicit without weakening channel isolation, bounded snapshots, or existing foreground/detached behavior. + +## Frozen success criteria + +### Gate A — Identity and history correctness (must pass) + +1. One stable `teamId` survives at least three rounds. +2. Every round has a unique `roundId` and monotonic `roundIndex`. +3. A round snapshot identifies both `teamId` and `roundId`. +4. The retained team exposes bounded summaries for completed/cancelled rounds; a new round does not overwrite the identity of an older one. +5. Existing member `sessionId` and `sessionRef` values remain stable across continuation rounds. + +### Gate B — Cancellation semantics (must pass) + +1. Cancelling a running round reaches a terminal round outcome within the existing cancellation timeout. +2. After cancellation, the team becomes available rather than permanently terminal. +3. `team_prompt` after cancellation starts a new round over the same member sessions. +4. Messages or completion from the cancelled round cannot mutate the new round. +5. Repeated cancel is deterministic and does not create an extra round. + +### Gate C — Backward compatibility (must pass) + +1. `team_start`, `team_get`, `team_wait`, `team_prompt`, and `team_cancel` remain usable through the existing retained handle. +2. The foreground manifest still provides one handle that can be passed to continuation tools. +3. Existing settlement meanings remain unchanged: `completed`, `quiescent`, and `exhausted` describe a round outcome, not objective correctness. +4. Existing public/direct/restricted-group routing and redaction tests remain green. +5. Existing bounded-output limits remain enforced. + +### Gate D — Observability (must pass) + +A snapshot must let an operator answer, without reading member transcripts: + +- Is the team available, running, closing, or closed? +- Which round is current/latest? +- What was that round's objective and terminal outcome? +- Was cancellation requested, and why? +- Which member sessions are retained? + +No restricted message body may be added to snapshots, round summaries, or events. + +### Gate E — Efficiency guardrails (must not regress) + +Use deterministic scripted agents; do not use live-model token cost as a CI gate. + +1. Starting a continuation creates no replacement member sessions. +2. Cancellation plus continuation adds no polling loop. +3. `team_wait` remains event-driven through monotonic `stateChangeSeq`. +4. Event history remains capped at the existing limit. +5. Round summaries must have an explicit fixed cap; exceeding it evicts oldest summaries without invalidating the stable team handle. + +## Required automated scenarios + +Create a focused lifecycle evaluation test suite covering: + +| Scenario | Required assertions | +|---|---| +| Three completed rounds | Stable `teamId`; three unique `roundId`s; indices 1, 2, 3; stable member sessions | +| Cancel then continue | Round 1 cancelled; team available; round 2 runs and settles normally | +| Cancel race | Late round-1 completion cannot settle or write into round 2 | +| Wait across transitions | `stateChangeSeq` strictly increases for cancel request, round settlement, and next-round start | +| History cap | Oldest round summary evicted at cap; latest/current identity remains correct | +| Privacy regression | Round metadata contains no direct/group plaintext canary | +| Existing channels | Public, DM, and restricted group tests remain unchanged and green | +| Foreground manifest | Stable team handle plus explicit latest round identity/outcome | + +## Baseline to record before implementation + +Run from repository root: + +```sh +npm test --workspace @geminixiang/pi-agent-team +npm run check --workspace @geminixiang/pi-agent-team +``` + +Record: + +- passing/failing test count; +- wall-clock test duration; +- current snapshot shape for start → settle → continue; +- current cancel → prompt behavior; +- current bounded event/result limits. + +A known pre-change failure that demonstrates the lifecycle ambiguity is allowed only in the new focused evaluation test. All pre-existing tests must remain green before implementation begins. + +## Acceptance command + +After implementation: + +```sh +npm test --workspace @geminixiang/pi-agent-team +npm run check --workspace @geminixiang/pi-agent-team +``` + +The change is accepted only when: + +1. Gates A–E pass in automated tests. +2. No pre-existing test is deleted or weakened to pass. +3. Public API changes are documented with a migration note. +4. A reviewer verifies that production code remains scenario-neutral. +5. The final diff contains no werewolf-specific production symbols. + +## Before / after evaluation table + +| Question | Before | Required after | +|---|---|---| +| What does the retained ID identify? | Team and latest run are conflated | Stable team handle only | +| Can a cancelled team continue? | Cancellation leaves an unusable retained record | A new round starts on the same team | +| Can two rounds be distinguished? | No first-class round identity/history | Unique `roundId`, index, bounded summaries | +| What does status describe? | Manager/run status mixes control plane and outcome | Team lifecycle and round lifecycle are separate | +| Are member histories retained? | Yes | Yes, with the same session identities | +| Does this add domain logic? | No | No; lifecycle remains rule-agnostic | + +## Non-goals for this change + +Do not combine these into the lifecycle patch: + +- werewolf/game state; +- durable recovery after parent-process restart; +- typed claim namespaces; +- block-reason confidentiality redesign; +- wait-abort UI wording; +- semantic/idempotent group aliases. + +Those require separate evaluations and diffs. Keeping them out makes lifecycle regression attribution possible. diff --git a/packages/pi-agent-team/src/extension.ts b/packages/pi-agent-team/src/extension.ts index f20f1aa..e938106 100644 --- a/packages/pi-agent-team/src/extension.ts +++ b/packages/pi-agent-team/src/extension.ts @@ -128,15 +128,16 @@ export default function agentTeam(pi: ExtensionAPI): void { params.members.map((member) => [member.id, new PiTeamAgent(member, ctx.cwd, ctx)]), ); let runtime!: TeamRuntime; + const roundId = crypto.randomUUID(); runtime = new TeamRuntime(params.objective, agents, { reactionDelayMs: { min: 50, max: 500 }, waitForIntervention: true, reporterId, reportPrompt: params.reportPrompt, - onActivity: (activity) => runs.observeActivity(runtime.teamId, activity), - onProgress: (progress) => runs.observeProgress(runtime.teamId, progress), + onActivity: (activity) => runs.observeActivity(runtime.teamId, activity, roundId), + onProgress: (progress) => runs.observeProgress(runtime.teamId, progress, roundId), }); - const snapshot = runs.start(runtime, initial); + const snapshot = runs.start(runtime, initial, { roundId }); return { content: [{ type: "text", text: renderRunSnapshot(snapshot) }], details: snapshot, @@ -198,10 +199,16 @@ export default function agentTeam(pi: ExtensionAPI): void { let retained = false; try { const result = await runtime.run(initial, signal); - runs.retain(runtime, result); + const retainedSnapshot = runs.retain(runtime, result); retained = true; return { - content: [{ type: "text", text: renderFinalContent(result, params.members) }], + content: [{ + type: "text", + text: renderFinalContent(result, params.members, { + roundId: retainedSnapshot.roundId, + roundIndex: retainedSnapshot.roundIndex, + }), + }], details: finalDetails(details(), result), }; } finally { @@ -315,21 +322,34 @@ function registerTeamCancel(pi: ExtensionAPI, runs: TeamRunManager): void { { runId: Type.String({ minLength: 1 }), reason: Type.Optional(Type.String({ minLength: 1, maxLength: 1_000 })), + roundId: Type.Optional( + Type.String({ + minLength: 1, + description: "Optional observed round identity; a stale retry becomes a no-op instead of cancelling a newer round.", + }), + ), }, { additionalProperties: false }, ), async execute(_id, params) { - return snapshotResult(runs.cancel(params.runId, params.reason)); + return snapshotResult(runs.cancel(params.runId, params.reason, params.roundId)); }, }); } function renderRunSnapshot(snapshot: TeamRunSnapshot): string { const lines = [ - `team run: ${snapshot.runId}`, - `status: ${snapshot.status}`, + `team: ${snapshot.teamId}`, + `lifecycle: ${snapshot.lifecycle}`, + `round: ${snapshot.roundId} (#${snapshot.roundIndex})`, + `objective: ${snapshot.objective}`, + `round status: ${snapshot.status}`, `stateChangeSeq: ${snapshot.stateChangeSeq}`, ]; + if (snapshot.cancellation) + lines.push( + `cancellation: requested at ${new Date(snapshot.cancellation.requestedAt).toISOString()} · ${snapshot.cancellation.reason}`, + ); if (snapshot.progress) lines.push( `progress: ${snapshot.progress.turns} turns · ${snapshot.progress.finished.length} finished · ${snapshot.progress.blocked.length} blocked`, @@ -364,6 +384,10 @@ function renderRunSnapshot(snapshot: TeamRunSnapshot): string { ); } if (snapshot.error) lines.push(`error: ${snapshot.error}`); + if (snapshot.rounds.length) + lines.push( + `round history: ${snapshot.rounds.map((round) => `#${round.roundIndex} ${round.roundId} ${round.status}`).join(" · ")}`, + ); const latest = snapshot.events.at(-1); if (latest) lines.push(`latest event: #${latest.sequence} ${latest.type} · ${latest.summary}`); return lines.join("\n"); @@ -384,6 +408,7 @@ function summarizeLive(details: TeamDisplayDetails): string { export function renderFinalContent( result: TeamResult, members: readonly { id: string; name: string }[], + round?: { roundId: string; roundIndex: number }, ): string { const nameOf = (id: string) => members.find((member) => member.id === id)?.name ?? id; const lines: string[] = []; @@ -405,6 +430,7 @@ export function renderFinalContent( lines.push( "TEAM MANIFEST", `team: ${result.teamId}`, + ...(round ? [`round: ${round.roundId} (#${round.roundIndex})`] : []), `settlement: ${result.settlement.kind} (${result.settlement.meaning}; objective correctness unverified)`, "members:", ); @@ -419,7 +445,7 @@ export function renderFinalContent( `messages: ${result.publicTranscript.length} public · ${result.restrictedMessages.length} restricted (bodies not included here)`, `audit: ${result.events.length} events · head ${result.auditHead}`, "Each member's full first-person history is in its session file listed above.", - `The team remains available in this parent session: use team_prompt with runId ${result.teamId} and a member id to start a continuation round.`, + `The team remains available in this parent session: use team_prompt with runId ${result.teamId} (the stable team handle) and a member id to start a continuation round.`, ); return lines.join("\n"); } diff --git a/packages/pi-agent-team/src/run-manager.ts b/packages/pi-agent-team/src/run-manager.ts index 922102c..5e5d1b7 100644 --- a/packages/pi-agent-team/src/run-manager.ts +++ b/packages/pi-agent-team/src/run-manager.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import type { MemberId, TeamActivity, @@ -9,17 +10,23 @@ import type { TeamInitialPost } from "./runtime.js"; export const MAX_TEAM_RUN_EVENTS = 512; export const MAX_TEAM_RUNS = 64; +/** Completed/cancelled round summaries retained per stable team handle. */ +export const MAX_TEAM_ROUND_SUMMARIES = 16; const MAX_REPORT_BYTES = 16 * 1024; const MAX_MEMBER_FIELD_BYTES = 256; const MAX_EVENT_SUMMARY_BYTES = 1_000; +const MAX_OBJECTIVE_BYTES = 8_000; export type TeamRunStatus = "running" | "cancelling" | "settled" | "cancelled" | "failed"; +export type TeamLifecycleStatus = "available" | "running" | "closing" | "closed"; export interface TeamRunEvent { sequence: number; at: number; type: "started" | "progress" | "activity" | "prompted" | "cancel-requested" | "settled" | "cancelled" | "failed"; summary: string; + roundId: string; + roundIndex: number; } export interface TeamRunResultSummary { @@ -42,25 +49,46 @@ export interface TeamRunResultSummary { userInterventions: number; } +export interface TeamRoundSummary { + teamId: string; + roundId: string; + roundIndex: number; + objective: string; + status: "settled" | "cancelled" | "failed"; + startedAt: number; + updatedAt: number; + cancellation?: { requested: true; requestedAt: number; reason: string }; + result?: TeamRunResultSummary; + error?: string; +} + +/** + * Snapshot of a stable retained team. `runId` remains as a compatibility + * alias for `teamId`; `roundId` identifies only the current/latest round. + */ export interface TeamRunSnapshot { runId: string; teamId: string; + lifecycle: TeamLifecycleStatus; + roundId: string; + roundIndex: number; + objective: string; status: TeamRunStatus; stateChangeSeq: number; startedAt: number; updatedAt: number; + cancellation?: { requested: true; requestedAt: number; reason: string }; progress?: TeamProgress; events: readonly TeamRunEvent[]; result?: TeamRunResultSummary; error?: string; + rounds: readonly TeamRoundSummary[]; } export interface ManagedTeamRuntime { readonly teamId: string; - run( - initial: TeamInitialPost | readonly TeamInitialPost[], - signal?: AbortSignal, - ): Promise; + readonly objective: string; + run(initial: TeamInitialPost | readonly TeamInitialPost[], signal?: AbortSignal): Promise; intervene(memberId: MemberId, message: string): void; hasMember(memberId: MemberId): boolean; next( @@ -77,272 +105,278 @@ export interface ManagedTeamRuntime { close?(): Promise | void; } -interface RunRecord { +interface TeamRecord { + teamId: string; runtime: ManagedTeamRuntime; controller: AbortController; status: TeamRunStatus; + roundId: string; + roundIndex: number; + objective: string; + generation: number; sequence: number; startedAt: number; updatedAt: number; + cancellation?: { requested: true; requestedAt: number; reason: string }; progress?: TeamProgress; events: TeamRunEvent[]; result?: TeamRunResultSummary; error?: string; + rounds: TeamRoundSummary[]; waiters: Set<() => void>; completion?: Promise; } export class TeamRunManager { - private readonly runs = new Map(); + private readonly teams = new Map(); start( runtime: ManagedTeamRuntime, initial: TeamInitialPost | readonly TeamInitialPost[], + options: { roundId?: string } = {}, ): TeamRunSnapshot { - const runId = runtime.teamId; - if (this.runs.has(runId)) throw new Error(`Team run already exists: ${runId}`); + const teamId = runtime.teamId; + if (this.teams.has(teamId)) throw new Error(`Team run already exists: ${teamId}`); this.makeRoom(); - const now = Date.now(); - const record: RunRecord = { - runtime, - controller: new AbortController(), - status: "running", - sequence: 0, - startedAt: now, - updatedAt: now, - events: [], - waiters: new Set(), - }; - this.runs.set(runId, record); - this.bump(record, "started", `team ${runtime.teamId} started`); + const record = this.newRecord(runtime, "running", options.roundId); + this.teams.set(teamId, record); + this.bump(record, "started", `team ${teamId} round 1 started`); this.launch(record, initial); return this.snapshot(record); } - /** Retain a synchronously completed team so later prompts can continue it. */ + /** Retain a synchronously completed first round for later continuation. */ retain(runtime: ManagedTeamRuntime, result: TeamResult): TeamRunSnapshot { - const runId = runtime.teamId; - if (this.runs.has(runId)) throw new Error(`Team run already exists: ${runId}`); + const teamId = runtime.teamId; + if (this.teams.has(teamId)) throw new Error(`Team run already exists: ${teamId}`); this.makeRoom(); - const now = Date.now(); - const record: RunRecord = { - runtime, - controller: new AbortController(), - status: "settled", - sequence: 0, - startedAt: now, - updatedAt: now, - events: [], - waiters: new Set(), - result: summarizeResult(result), - completion: Promise.resolve(), - }; - this.runs.set(runId, record); + const record = this.newRecord(runtime, "settled"); + record.result = summarizeResult(result); + record.completion = Promise.resolve(); + this.teams.set(teamId, record); this.bump(record, "settled", `${result.settlement.kind} (${result.settlement.meaning})`); + this.archiveRound(record); return this.snapshot(record); } get(runId: string): TeamRunSnapshot { - return this.snapshot(this.requireRun(runId)); + return this.snapshot(this.requireTeam(runId)); } - observeProgress(runId: string, progress: TeamProgress): void { - const record = this.requireActive(runId); + observeProgress(runId: string, progress: TeamProgress, roundId?: string): void { + const record = this.requireTeam(runId); + if (roundId !== undefined && record.roundId !== roundId) return; + if (record.status !== "running") return; record.progress = progress; - this.bump( - record, - "progress", - `${progress.turns} turns; ${progress.finished.length} finished; ${progress.blocked.length} blocked`, - ); + this.bump(record, "progress", `${progress.turns} turns; ${progress.finished.length} finished; ${progress.blocked.length} blocked`); } - observeActivity(runId: string, activity: TeamActivity): void { - const record = this.requireActive(runId); + observeActivity(runId: string, activity: TeamActivity, roundId?: string): void { + const record = this.requireTeam(runId); + if (roundId !== undefined && record.roundId !== roundId) return; + if (record.status !== "running") return; this.bump(record, "activity", `${activity.memberId} ${activity.kind}`); } prompt(runId: string, memberId: MemberId, message: string): TeamRunSnapshot { - if (!message.trim()) throw new Error("Team prompt message must not be empty"); - if (Buffer.byteLength(message, "utf8") > 8_000) - throw new Error("Team prompt message must not exceed 8000 bytes"); - const record = this.requireRun(runId); + if (!message.trim()) throw new Error("Prompt message must not be empty"); + if (Buffer.byteLength(message, "utf8") > MAX_OBJECTIVE_BYTES) + throw new Error(`Prompt message exceeds ${MAX_OBJECTIVE_BYTES} UTF-8 bytes`); + const record = this.requireTeam(runId); if (record.status === "running") { record.runtime.intervene(memberId, message); this.bump(record, "prompted", `prompted member ${memberId}`); return this.snapshot(record); } - if (record.status !== "settled") - throw new Error(`Team run ${runId} is ${record.status} and no longer accepts prompts`); + if (record.status === "cancelling") + throw new Error(`Team run ${runId} is cancelling and no longer accepts mutations`); if (!record.runtime.hasMember(memberId)) throw new Error(`Unknown member: ${memberId}`); - record.runtime = record.runtime.next( + const nextRoundId = randomUUID(); + const next = record.runtime.next( message, { - waitForIntervention: true, + waitForIntervention: false, reporterId: memberId, - reportPrompt: - "Reply directly to the requester about this continuation objective. Include relevant work or teammate results, and make the response stand alone.", - onActivity: (activity) => this.observeActivity(runId, activity), - onProgress: (progress) => this.observeProgress(runId, progress), + reportPrompt: "Reply directly to the requester guidance that started this continuation round. Return a complete, standalone response.", + onActivity: (activity) => this.observeActivity(record.teamId, activity, nextRoundId), + onProgress: (progress) => this.observeProgress(record.teamId, progress, nextRoundId), }, true, ); + record.runtime = next; record.controller = new AbortController(); record.status = "running"; + record.roundId = nextRoundId; + record.roundIndex += 1; + record.objective = `Requester continuation directed to member ${memberId}`; + record.generation += 1; + record.startedAt = Date.now(); + record.updatedAt = record.startedAt; + record.cancellation = undefined; record.progress = undefined; record.result = undefined; record.error = undefined; - this.bump(record, "prompted", `continued team by prompting member ${memberId}`); + this.bump(record, "started", `continued team by prompting member ${memberId}`); this.launch(record, { channel: { kind: "direct", memberId }, body: message }); return this.snapshot(record); } - cancel(runId: string, reason = "cancelled by requester"): TeamRunSnapshot { + cancel(runId: string, reason = "cancelled by requester", expectedRoundId?: string): TeamRunSnapshot { + const record = this.requireTeam(runId); + if (expectedRoundId !== undefined && expectedRoundId !== record.roundId) return this.snapshot(record); if (!reason.trim()) throw new Error("Cancellation reason must not be empty"); - if (Buffer.byteLength(reason, "utf8") > 1_000) - throw new Error("Cancellation reason must not exceed 1000 bytes"); - const record = this.requireRunning(runId); + if (record.status === "cancelling" || record.status === "cancelled") return this.snapshot(record); + if (record.status !== "running") + throw new Error(`Team run ${runId} is ${record.status} and no longer active`); record.status = "cancelling"; - this.bump(record, "cancel-requested", reason); - record.controller.abort(new Error(reason)); + record.cancellation = Object.freeze({ requested: true, requestedAt: Date.now(), reason: truncateUtf8(reason, MAX_EVENT_SUMMARY_BYTES) }); + this.bump(record, "cancel-requested", record.cancellation.reason); + record.controller.abort(new Error(record.cancellation.reason)); return this.snapshot(record); } - cancelAll(reason = "parent session shut down"): void { - for (const [runId, record] of this.runs) - if (record.status === "running") this.cancel(runId, reason); - } - - async shutdown(reason = "parent session shut down"): Promise { - this.cancelAll(reason); - await Promise.all([...this.runs.values()].map((record) => record.completion)); - await Promise.allSettled([...this.runs.values()].map((record) => record.runtime.close?.())); - } - - async wait( - runId: string, - options: { afterSeq?: number; timeoutMs?: number; signal?: AbortSignal } = {}, - ): Promise { - const record = this.requireRun(runId); + async wait(runId: string, options: { afterSeq?: number; timeoutMs?: number; signal?: AbortSignal } = {}): Promise { + const record = this.requireTeam(runId); const afterSeq = options.afterSeq ?? record.sequence; if (!Number.isInteger(afterSeq) || afterSeq < 0) throw new Error("afterSeq must be a non-negative integer"); if (afterSeq > record.sequence) throw new Error(`afterSeq ${afterSeq} is ahead of current stateChangeSeq ${record.sequence}`); if (record.sequence > afterSeq || terminal(record.status)) return this.snapshot(record); - await this.waitForChange(record, options.timeoutMs, options.signal); - return this.snapshot(record); - } - - private async waitForChange( - record: RunRecord, - timeoutMs?: number, - signal?: AbortSignal, - ): Promise { - if (timeoutMs !== undefined && (!Number.isFinite(timeoutMs) || timeoutMs <= 0)) - throw new Error("timeoutMs must be greater than zero"); - if (signal?.aborted) throw signal.reason ?? new Error("Team wait aborted"); await new Promise((resolve, reject) => { let timer: ReturnType | undefined; const cleanup = () => { - record.waiters.delete(onChange); - signal?.removeEventListener("abort", onAbort); + record.waiters.delete(wake); if (timer) clearTimeout(timer); + options.signal?.removeEventListener("abort", abort); }; - const onChange = () => { - cleanup(); - resolve(); - }; - const onAbort = () => { - cleanup(); - reject(signal?.reason ?? new Error("Team wait aborted")); - }; - record.waiters.add(onChange); - signal?.addEventListener("abort", onAbort, { once: true }); - if (timeoutMs !== undefined) - timer = setTimeout(() => { - cleanup(); - reject(new Error(`Timed out after ${timeoutMs}ms waiting for team run ${record.runtime.teamId}`)); - }, timeoutMs); + const wake = () => { cleanup(); resolve(); }; + const abort = () => { cleanup(); reject(options.signal?.reason ?? new Error("Wait aborted")); }; + record.waiters.add(wake); + if (options.timeoutMs !== undefined) + timer = setTimeout(() => { cleanup(); reject(new Error(`Timed out waiting for team run ${runId}`)); }, options.timeoutMs); + options.signal?.addEventListener("abort", abort, { once: true }); + if (options.signal?.aborted) abort(); + else if (record.sequence > afterSeq) wake(); }); + return this.snapshot(record); } - private launch( - record: RunRecord, - initial: TeamInitialPost | readonly TeamInitialPost[], - ): void { - record.completion = record.runtime.run(initial, record.controller.signal).then( + async shutdown(reason = "parent session ended"): Promise { + const completions: Promise[] = []; + for (const record of this.teams.values()) { + if (record.status === "running") this.cancel(record.teamId, reason); + if (record.completion) completions.push(record.completion); + } + await Promise.allSettled(completions); + await Promise.allSettled([...this.teams.values()].map((record) => record.runtime.close?.())); + this.teams.clear(); + } + + private newRecord(runtime: ManagedTeamRuntime, status: TeamRunStatus, roundId: string = randomUUID()): TeamRecord { + const now = Date.now(); + return { + teamId: runtime.teamId, runtime, controller: new AbortController(), status, + roundId, roundIndex: 1, objective: runtime.objective, + generation: 1, sequence: 0, startedAt: now, updatedAt: now, + events: [], rounds: [], waiters: new Set(), + }; + } + + private launch(record: TeamRecord, initial: TeamInitialPost | readonly TeamInitialPost[]): void { + const generation = record.generation; + const runtime = record.runtime; + const controller = record.controller; + record.completion = runtime.run(initial, controller.signal).then( (result) => { + if (record.generation !== generation || record.runtime !== runtime) return; + if (controller.signal.aborted) { + record.status = "cancelled"; + record.error = truncateUtf8(errorMessage(controller.signal.reason), MAX_EVENT_SUMMARY_BYTES); + this.bump(record, "cancelled", record.error); + this.archiveRound(record); + return; + } record.status = "settled"; record.result = summarizeResult(result); this.bump(record, "settled", `${result.settlement.kind} (${result.settlement.meaning})`); + this.archiveRound(record); }, (cause) => { + if (record.generation !== generation || record.runtime !== runtime) return; const message = truncateUtf8(errorMessage(cause), MAX_EVENT_SUMMARY_BYTES); record.error = message; - record.status = record.controller.signal.aborted ? "cancelled" : "failed"; + record.status = controller.signal.aborted ? "cancelled" : "failed"; this.bump(record, record.status, message); + this.archiveRound(record); }, ); } - private bump(record: RunRecord, type: TeamRunEvent["type"], summary: string): void { + private archiveRound(record: TeamRecord): void { + const summary: TeamRoundSummary = Object.freeze({ + teamId: record.teamId, + roundId: record.roundId, + roundIndex: record.roundIndex, + objective: truncateUtf8(record.objective, MAX_OBJECTIVE_BYTES), + status: record.status as TeamRoundSummary["status"], + startedAt: record.startedAt, + updatedAt: record.updatedAt, + cancellation: record.cancellation, + result: record.result, + error: record.error, + }); + record.rounds.push(summary); + if (record.rounds.length > MAX_TEAM_ROUND_SUMMARIES) + record.rounds.splice(0, record.rounds.length - MAX_TEAM_ROUND_SUMMARIES); + } + + private bump(record: TeamRecord, type: TeamRunEvent["type"], summary: string): void { record.sequence += 1; record.updatedAt = Date.now(); - record.events.push({ - sequence: record.sequence, - at: record.updatedAt, - type, - summary: truncateUtf8(summary, MAX_EVENT_SUMMARY_BYTES), - }); + record.events.push({ sequence: record.sequence, at: record.updatedAt, type, summary: truncateUtf8(summary, MAX_EVENT_SUMMARY_BYTES), roundId: record.roundId, roundIndex: record.roundIndex }); if (record.events.length > MAX_TEAM_RUN_EVENTS) record.events.splice(0, record.events.length - MAX_TEAM_RUN_EVENTS); record.waiters.forEach((wake) => wake()); } private makeRoom(): void { - if (this.runs.size < MAX_TEAM_RUNS) return; - const oldestTerminal = [...this.runs.entries()] + if (this.teams.size < MAX_TEAM_RUNS) return; + const oldestAvailable = [...this.teams.entries()] .filter(([, record]) => terminal(record.status)) .sort(([, left], [, right]) => left.updatedAt - right.updatedAt)[0]; - if (!oldestTerminal) - throw new Error(`Too many active team runs; maximum is ${MAX_TEAM_RUNS}`); - this.runs.delete(oldestTerminal[0]); - void Promise.resolve(oldestTerminal[1].runtime.close?.()).catch(() => {}); + if (!oldestAvailable) throw new Error(`Too many active team runs; maximum is ${MAX_TEAM_RUNS}`); + this.teams.delete(oldestAvailable[0]); + void Promise.resolve(oldestAvailable[1].runtime.close?.()).catch(() => {}); } - private requireRun(runId: string): RunRecord { - const record = this.runs.get(runId); + private requireTeam(runId: string): TeamRecord { + const record = this.teams.get(runId); if (!record) throw new Error(`Unknown team run: ${runId}`); return record; } - private requireRunning(runId: string): RunRecord { - const record = this.requireRun(runId); - if (record.status !== "running") - throw new Error(`Team run ${runId} is ${record.status} and no longer accepts mutations`); - return record; - } - - private requireActive(runId: string): RunRecord { - const record = this.requireRun(runId); - if (terminal(record.status)) - throw new Error(`Team run ${runId} is ${record.status} and no longer active`); - return record; - } - private snapshot(record: RunRecord): TeamRunSnapshot { + private snapshot(record: TeamRecord): TeamRunSnapshot { + const lifecycle: TeamLifecycleStatus = record.status === "running" ? "running" : record.status === "cancelling" ? "closing" : "available"; return Object.freeze({ - runId: record.runtime.teamId, - teamId: record.runtime.teamId, + runId: record.teamId, + teamId: record.teamId, + lifecycle, + roundId: record.roundId, + roundIndex: record.roundIndex, + objective: truncateUtf8(record.objective, MAX_OBJECTIVE_BYTES), status: record.status, stateChangeSeq: record.sequence, startedAt: record.startedAt, updatedAt: record.updatedAt, + cancellation: record.cancellation, progress: record.progress, events: Object.freeze(record.events.map((event) => Object.freeze({ ...event }))), result: record.result, error: record.error, + rounds: Object.freeze([...record.rounds]), }); } } @@ -355,31 +389,16 @@ function summarizeResult(result: TeamResult): TeamRunResultSummary { return Object.freeze({ settlement: result.settlement, objectiveVerification: result.objectiveVerification, - report: result.report - ? Object.freeze({ - reporterId: result.report.reporterId, - body: truncateUtf8(result.report.body, MAX_REPORT_BYTES), - }) - : undefined, + report: result.report ? Object.freeze({ reporterId: result.report.reporterId, body: truncateUtf8(result.report.body, MAX_REPORT_BYTES) }) : undefined, reportError: truncateOptional(result.reportError, MAX_EVENT_SUMMARY_BYTES), - members: Object.freeze( - result.members.map((member) => - Object.freeze({ - id: truncateUtf8(member.id, 128), - sessionId: truncateUtf8(member.sessionId, 128), - sessionRef: truncateOptional(member.sessionRef, 512), - turns: member.turns, - state: member.state, - summary: truncateOptional(member.summary, MAX_MEMBER_FIELD_BYTES), - error: truncateOptional(member.error, MAX_MEMBER_FIELD_BYTES), - blockedReason: truncateOptional(member.blockedReason, MAX_MEMBER_FIELD_BYTES), - }), - ), - ), - messageCounts: Object.freeze({ - public: result.publicTranscript.length, - restricted: result.restrictedMessages.length, - }), + members: Object.freeze(result.members.map((member) => Object.freeze({ + id: truncateUtf8(member.id, 128), sessionId: truncateUtf8(member.sessionId, 128), + sessionRef: truncateOptional(member.sessionRef, 512), turns: member.turns, state: member.state, + summary: truncateOptional(member.summary, MAX_MEMBER_FIELD_BYTES), + error: truncateOptional(member.error, MAX_MEMBER_FIELD_BYTES), + blockedReason: truncateOptional(member.blockedReason, MAX_MEMBER_FIELD_BYTES), + }))), + messageCounts: Object.freeze({ public: result.publicTranscript.length, restricted: result.restrictedMessages.length }), audit: Object.freeze({ events: result.events.length, head: result.auditHead }), userInterventions: result.userInterventions, }); diff --git a/packages/pi-agent-team/test/run-manager.test.ts b/packages/pi-agent-team/test/run-manager.test.ts index fd6d908..31c932f 100644 --- a/packages/pi-agent-team/test/run-manager.test.ts +++ b/packages/pi-agent-team/test/run-manager.test.ts @@ -2,6 +2,7 @@ import assert from "node:assert/strict"; import test from "node:test"; import type { TeamActivity, TeamProgress, TeamResult } from "../src/domain.js"; import { + MAX_TEAM_ROUND_SUMMARIES, MAX_TEAM_RUN_EVENTS, TeamRunManager, type ManagedTeamRuntime, @@ -104,12 +105,68 @@ const activity: TeamActivity = { targetIds: ["a"], }; +test("three rounds retain stable identity, sessions, and monotonic round identities", async () => { + const manager = new TeamRunManager(); + const runtime = new FakeRuntime("stable-team"); + const first = manager.start(runtime, { channel: { kind: "public" }, body: "one" }); + runtime.pending.resolve(result("stable-team", "one")); + const one = await manager.wait("stable-team", { afterSeq: first.stateChangeSeq, timeoutMs: 1_000 }); + const twoStarted = manager.prompt("stable-team", "a", "two"); + runtime.continuation!.pending.resolve(result("stable-team", "two")); + const two = await manager.wait("stable-team", { afterSeq: twoStarted.stateChangeSeq, timeoutMs: 1_000 }); + const threeStarted = manager.prompt("stable-team", "a", "three"); + runtime.continuation!.continuation!.pending.resolve(result("stable-team", "three")); + const three = await manager.wait("stable-team", { afterSeq: threeStarted.stateChangeSeq, timeoutMs: 1_000 }); + + assert.equal(three.teamId, first.teamId); + assert.deepEqual(three.rounds.map((round) => round.roundIndex), [1, 2, 3]); + assert.equal(new Set(three.rounds.map((round) => round.roundId)).size, 3); + assert.deepEqual(three.rounds.map((round) => round.result?.members[0].sessionId), ["session-a", "session-a", "session-a"]); + assert.ok(two.stateChangeSeq < threeStarted.stateChangeSeq); +}); + +test("round history is capped without invalidating the stable handle", async () => { + const manager = new TeamRunManager(); + let runtime = new FakeRuntime("history-team"); + let snapshot = manager.start(runtime, { channel: { kind: "public" }, body: "one" }); + runtime.pending.resolve(result("history-team")); + snapshot = await manager.wait("history-team", { afterSeq: snapshot.stateChangeSeq, timeoutMs: 1_000 }); + for (let index = 2; index <= MAX_TEAM_ROUND_SUMMARIES + 2; index++) { + snapshot = manager.prompt("history-team", "a", `round ${index}`); + runtime = runtime.continuation!; + runtime.pending.resolve(result("history-team")); + snapshot = await manager.wait("history-team", { afterSeq: snapshot.stateChangeSeq, timeoutMs: 1_000 }); + } + assert.equal(snapshot.rounds.length, MAX_TEAM_ROUND_SUMMARIES); + assert.equal(snapshot.rounds[0].roundIndex, 3); + assert.equal(snapshot.rounds.at(-1)?.roundId, snapshot.roundId); + assert.equal(manager.get("history-team").teamId, "history-team"); +}); + +test("late callbacks and completion from a cancelled round cannot mutate its continuation", async () => { + class LateRuntime extends FakeRuntime { + override run(): Promise { return this.pending.promise; } + } + const manager = new TeamRunManager(); + const runtime = new LateRuntime("race-team"); + const first = manager.start(runtime, { channel: { kind: "public" }, body: "one" }); + manager.cancel("race-team", "stop"); + runtime.pending.reject(new Error("cancelled late")); + await manager.wait("race-team", { afterSeq: first.stateChangeSeq, timeoutMs: 1_000 }); + const second = manager.prompt("race-team", "a", "two"); + manager.observeProgress("race-team", progress("race-team"), first.roundId); + assert.equal(manager.get("race-team").progress, undefined); + assert.equal(manager.get("race-team").roundId, second.roundId); +}); test("detached runs return immediately, wait by sequence without polling, and expose a bounded result", async () => { const manager = new TeamRunManager(); const runtime = new FakeRuntime("run-1"); const started = manager.start(runtime, { channel: { kind: "public" }, body: "start" }); assert.equal(started.status, "running"); assert.equal(started.runId, "run-1"); + assert.equal(started.teamId, "run-1"); + assert.equal(started.roundIndex, 1); + assert.notEqual(started.roundId, started.teamId); const waiting = manager.wait("run-1", { afterSeq: started.stateChangeSeq, timeoutMs: 1_000 }); manager.observeProgress("run-1", progress("run-1")); @@ -143,6 +200,9 @@ test("a settled team accepts a new prompt as a fresh round over the same handle" const continued = manager.prompt("run-continue", "a", "do a different task"); assert.equal(continued.status, "running"); assert.equal(continued.runId, "run-continue"); + assert.equal(continued.teamId, settled.teamId); + assert.equal(continued.roundIndex, 2); + assert.notEqual(continued.roundId, settled.roundId); assert.equal(runtime.continuation?.objective, "do a different task"); assert.equal(runtime.continuation?.continuationOptions?.reporterId, "a"); assert.match(runtime.continuation?.continuationOptions?.reportPrompt ?? "", /Reply directly/); @@ -156,6 +216,7 @@ test("a settled team accepts a new prompt as a fresh round over the same handle" }); assert.equal(second.status, "settled"); assert.equal(second.result?.report?.body, "second done"); + assert.deepEqual(second.rounds.map((round) => round.roundIndex), [1, 2]); assert.ok(second.events.some((event) => event.summary === "continued team by prompting member a")); }); @@ -197,7 +258,7 @@ test("event history is capped and activity bodies are reduced to lightweight sum assert.equal(snapshot.events.at(-1)?.summary, "a wake"); }); -test("cancel has its own signal, wakes waiters, and terminal or invalid mutations fail explicitly", async () => { +test("cancel has its own signal, wakes waiters, is idempotent, and permits continuation", async () => { const manager = new TeamRunManager(); const runtime = new FakeRuntime("run-cancel"); const started = manager.start(runtime, { channel: { kind: "public" }, body: "start" }); @@ -212,10 +273,15 @@ test("cancel has its own signal, wakes waiters, and terminal or invalid mutation ); await new Promise((resolve) => setImmediate(resolve)); - assert.equal(manager.get("run-cancel").status, "cancelled"); - assert.throws(() => manager.prompt("run-cancel", "a", "resume"), /no longer accepts/); - assert.throws(() => manager.cancel("run-cancel"), /no longer accepts/); - assert.throws(() => manager.get("missing"), /Unknown team run/); + const cancelled = manager.get("run-cancel"); + assert.equal(cancelled.status, "cancelled"); + assert.equal(cancelled.lifecycle, "available"); + assert.equal(manager.cancel("run-cancel").roundId, cancelled.roundId); + const continued = manager.prompt("run-cancel", "a", "resume"); + assert.equal(continued.status, "running"); + assert.equal(continued.roundIndex, 2); + assert.notEqual(continued.roundId, cancelled.roundId); + assert.throws(() => manager.get("missing"), /Unknown team/); }); test("wait times out and caller abort does not cancel the detached run", async () => { From e1bffe487520f380590df0e5fdbb796febac2b4e Mon Sep 17 00:00:00 2001 From: Ying Xiang Date: Thu, 20 Aug 2026 12:50:58 +0800 Subject: [PATCH 2/3] test(pi-agent-team): add lifecycle evaluation coverage --- .../test/lifecycle-evaluation.test.ts | 317 ++++++++++++++++++ 1 file changed, 317 insertions(+) create mode 100644 packages/pi-agent-team/test/lifecycle-evaluation.test.ts diff --git a/packages/pi-agent-team/test/lifecycle-evaluation.test.ts b/packages/pi-agent-team/test/lifecycle-evaluation.test.ts new file mode 100644 index 0000000..5fc20f6 --- /dev/null +++ b/packages/pi-agent-team/test/lifecycle-evaluation.test.ts @@ -0,0 +1,317 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import type { TeamActivity, TeamResult } from "../src/domain.js"; +import { renderFinalContent } from "../src/extension.js"; +import * as retainedApi from "../src/run-manager.js"; +import { + TeamRunManager, + type ManagedTeamRuntime, + type TeamRunSnapshot, +} from "../src/run-manager.js"; + +/** + * Black-box lifecycle evaluation for docs/lifecycle-improvement-evaluation.md. + * + * These adapters deliberately accept either a nested `latestRound`/`currentRound` + * TeamHandle representation or an equivalent flat latest-round representation. + * Assertions below concern lifecycle semantics, not incidental property nesting. + */ +type EvaluationRound = { + roundId?: string; + roundIndex?: number; + objective?: string; + status?: string; + outcome?: unknown; + result?: unknown; + cancellation?: { requested?: boolean; reason?: string }; + cancellationRequested?: boolean; + cancellationReason?: string; +}; + +type EvaluationHandle = TeamRunSnapshot & { + teamStatus?: string; + currentRound?: EvaluationRound; + latestRound?: EvaluationRound; + round?: EvaluationRound; + rounds?: readonly EvaluationRound[]; + roundHistory?: readonly EvaluationRound[]; + members?: readonly { id: string; sessionId: string; sessionRef?: string }[]; +}; + +function handle(snapshot: TeamRunSnapshot): EvaluationHandle { + return snapshot as EvaluationHandle; +} + +function latest(snapshot: TeamRunSnapshot): EvaluationRound { + const value = handle(snapshot).latestRound ?? handle(snapshot).currentRound ?? handle(snapshot).round; + assert.ok(value, "retained TeamHandle snapshot must expose explicit latest/current RoundRun identity"); + return value; +} + +function history(snapshot: TeamRunSnapshot): readonly EvaluationRound[] { + const value = handle(snapshot).roundHistory ?? handle(snapshot).rounds; + assert.ok(value, "retained TeamHandle snapshot must expose bounded round summaries"); + return value; +} + +function roundOutcome(round: EvaluationRound): unknown { + return round.outcome ?? round.result ?? round.status; +} + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (cause: unknown) => void; + const promise = new Promise((yes, no) => { + resolve = yes; + reject = no; + }); + return { promise, resolve, reject }; +} + +const sessions = Object.freeze([ + { id: "a", sessionId: "session-a", sessionRef: "/sessions/a.jsonl" }, + { id: "b", sessionId: "session-b", sessionRef: "/sessions/b.jsonl" }, +]); + +function completedResult(teamId: string, label: string): TeamResult { + return { + teamId, + settlement: { kind: "completed", meaning: "all-members-finished" }, + objectiveVerification: "unverified", + report: { reporterId: "a", body: `${label} complete` }, + members: sessions.map((member) => ({ ...member, turns: 1, state: "finished", summary: label })), + publicTranscript: [], + restrictedMessages: [], + events: [], + userInterventions: 0, + auditHead: "0".repeat(64), + }; +} + +class ScriptedRound implements ManagedTeamRuntime { + readonly pending = deferred(); + readonly aborts: unknown[] = []; + continuation?: ScriptedRound; + + constructor( + readonly teamId: string, + readonly objective: string, + private readonly abortMode: "reject" | "ignore" = "reject", + ) {} + + run(_initial: unknown, signal?: AbortSignal): Promise { + signal?.addEventListener("abort", () => { + this.aborts.push(signal.reason); + if (this.abortMode === "reject") this.pending.reject(signal.reason ?? new Error("aborted")); + }, { once: true }); + return this.pending.promise; + } + + intervene(): void {} + + hasMember(memberId: string): boolean { + return sessions.some((member) => member.id === memberId); + } + + next(objective: string): ScriptedRound { + return (this.continuation = new ScriptedRound(this.teamId, objective)); + } +} + +async function settle( + manager: TeamRunManager, + runtime: ScriptedRound, + label: string, +): Promise { + const before = manager.get(runtime.teamId).stateChangeSeq; + runtime.pending.resolve(completedResult(runtime.teamId, label)); + return manager.wait(runtime.teamId, { afterSeq: before, timeoutMs: 1_000 }); +} + +async function nextTick(): Promise { + await new Promise((resolve) => setImmediate(resolve)); +} + +test("lifecycle evaluation: three rounds retain team and member identity with distinct history", async () => { + const manager = new TeamRunManager(); + const first = new ScriptedRound("team-three", "objective one"); + manager.start(first, { channel: { kind: "public" }, body: "begin" }); + await settle(manager, first, "one"); + + manager.prompt(first.teamId, "a", "objective two"); + const second = first.continuation!; + await settle(manager, second, "two"); + + manager.prompt(first.teamId, "a", "objective three"); + const third = second.continuation!; + const final = await settle(manager, third, "three"); + const rounds = history(final); + + assert.equal(final.teamId, "team-three"); + assert.deepEqual(rounds.map((round) => round.roundIndex), [1, 2, 3]); + assert.equal(new Set(rounds.map((round) => round.roundId)).size, 3); + assert.ok(rounds.every((round) => typeof round.roundId === "string" && round.roundId.length > 0)); + assert.equal(latest(final).roundId, rounds[2].roundId); + assert.deepEqual(rounds.map((round) => round.objective), ["objective one", "objective two", "objective three"]); + + const retainedMembers = handle(final).members ?? (latest(final).result as TeamResult | undefined)?.members; + assert.deepEqual( + retainedMembers?.map(({ id, sessionId, sessionRef }) => ({ id, sessionId, sessionRef })), + sessions, + "continuations must retain the original member sessions rather than create replacements", + ); +}); + +test("lifecycle evaluation: cancel makes the team available and continuation settles", async () => { + const manager = new TeamRunManager(); + const first = new ScriptedRound("team-cancel", "cancel me"); + manager.start(first, { channel: { kind: "public" }, body: "begin" }); + manager.cancel(first.teamId, "requester changed direction"); + await nextTick(); + + const cancelled = manager.get(first.teamId); + assert.equal(handle(cancelled).teamStatus ?? cancelled.status, "available"); + assert.match(JSON.stringify(latest(cancelled)), /requester changed direction/); + assert.match(JSON.stringify(roundOutcome(latest(cancelled))), /cancel/i); + + const continued = manager.prompt(first.teamId, "a", "replacement objective"); + assert.equal(handle(continued).teamStatus ?? continued.status, "running"); + const second = first.continuation!; + const settled = await settle(manager, second, "replacement"); + assert.equal(settled.teamId, first.teamId); + assert.equal(latest(settled).roundIndex, 2); + assert.match(JSON.stringify(roundOutcome(latest(settled))), /completed|settled/); + assert.deepEqual(second.aborts, []); +}); + +test("lifecycle evaluation: cancelled-round completion cannot mutate its successor", async () => { + const manager = new TeamRunManager(); + const first = new ScriptedRound("team-race", "stale objective", "ignore"); + manager.start(first, { channel: { kind: "public" }, body: "begin" }); + manager.cancel(first.teamId, "replace round"); + + // The lifecycle contract must terminalize cancellation independently of an + // uncooperative old promise, allowing the retained team to continue. + await nextTick(); + const cancelledRoundId = latest(manager.get(first.teamId)).roundId; + manager.prompt(first.teamId, "a", "live objective"); + const second = first.continuation!; + const liveRoundId = latest(manager.get(first.teamId)).roundId; + assert.notEqual(liveRoundId, cancelledRoundId); + + first.pending.resolve(completedResult(first.teamId, "STALE-ROUND-CANARY")); + await nextTick(); + const afterLateCompletion = manager.get(first.teamId); + assert.equal(latest(afterLateCompletion).roundId, liveRoundId); + assert.equal(JSON.stringify(latest(afterLateCompletion)).includes("STALE-ROUND-CANARY"), false); + + const settled = await settle(manager, second, "live"); + assert.equal(latest(settled).roundId, liveRoundId); + assert.match(JSON.stringify(roundOutcome(latest(settled))), /completed|settled/); +}); + +test("lifecycle evaluation: wait observes strictly monotonic cancel, settlement, and restart", async () => { + const manager = new TeamRunManager(); + const first = new ScriptedRound("team-sequence", "first"); + const started = manager.start(first, { channel: { kind: "public" }, body: "begin" }); + + const cancelWait = manager.wait(first.teamId, { afterSeq: started.stateChangeSeq, timeoutMs: 1_000 }); + manager.cancel(first.teamId, "stop"); + const cancelRequested = await cancelWait; + + const terminal = await manager.wait(first.teamId, { + afterSeq: cancelRequested.stateChangeSeq, + timeoutMs: 1_000, + }); + const restart = manager.prompt(first.teamId, "a", "second"); + + assert.ok(started.stateChangeSeq < cancelRequested.stateChangeSeq); + assert.ok(cancelRequested.stateChangeSeq < terminal.stateChangeSeq); + assert.ok(terminal.stateChangeSeq < restart.stateChangeSeq); + assert.equal(latest(restart).roundIndex, 2); + + const repeated = manager.cancel(first.teamId, "stop second"); + const repeatedAgain = manager.cancel(first.teamId, "stop second"); + assert.equal(latest(repeatedAgain).roundId, latest(repeated).roundId, "repeated cancel creates no round"); +}); + +test("lifecycle evaluation: round summaries have an explicit cap and evict oldest first", async () => { + const exportedCap = (retainedApi as Record).MAX_TEAM_ROUND_SUMMARIES; + assert.equal(typeof exportedCap, "number", "round-summary history must publish an explicit fixed cap"); + const cap = exportedCap as number; + assert.ok(Number.isInteger(cap) && cap > 0 && cap <= 512); + + const manager = new TeamRunManager(); + let runtime = new ScriptedRound("team-history-cap", "objective 1"); + manager.start(runtime, { channel: { kind: "public" }, body: "begin" }); + await settle(manager, runtime, "round 1"); + for (let index = 2; index <= cap + 1; index++) { + manager.prompt(runtime.teamId, "a", `objective ${index}`); + runtime = runtime.continuation!; + await settle(manager, runtime, `round ${index}`); + } + + const snapshot = manager.get(runtime.teamId); + const rounds = history(snapshot); + assert.equal(rounds.length, cap); + assert.deepEqual(rounds.map((round) => round.roundIndex), Array.from({ length: cap }, (_, i) => i + 2)); + assert.equal(latest(snapshot).roundIndex, cap + 1); + assert.equal(snapshot.teamId, "team-history-cap"); +}); + +test("lifecycle evaluation: round metadata never leaks direct or restricted-group plaintext", () => { + const manager = new TeamRunManager(); + const runtime = new ScriptedRound("team-private", "safe objective"); + manager.start(runtime, { channel: { kind: "public" }, body: "begin" }); + + const directCanary = "DIRECT-PLAINTEXT-CANARY-7df08a"; + const groupCanary = "GROUP-PLAINTEXT-CANARY-45c991"; + const activities: TeamActivity[] = [ + { + sequence: 1, + memberId: "a", + kind: "message", + text: directCanary, + body: directCanary, + visibility: "restricted", + channel: { kind: "direct", memberId: "b" }, + targetIds: ["b"], + }, + { + sequence: 2, + memberId: "a", + kind: "message", + text: groupCanary, + body: groupCanary, + visibility: "restricted", + channel: { kind: "group", channelId: "secret" }, + targetIds: ["b"], + }, + ]; + for (const activity of activities) manager.observeActivity(runtime.teamId, activity); + + const serialized = JSON.stringify(manager.get(runtime.teamId)); + assert.equal(serialized.includes(directCanary), false); + assert.equal(serialized.includes(groupCanary), false); +}); + +test("lifecycle evaluation: foreground manifest names stable handle and latest round outcome", () => { + const foregroundResult = { + ...completedResult("team-foreground", "foreground"), + roundId: "round-3", + roundIndex: 3, + objective: "third objective", + } as TeamResult; + const manifest = renderFinalContent(foregroundResult, [ + { id: "a", name: "Agent A" }, + { id: "b", name: "Agent B" }, + ]); + + assert.match(manifest, /team(?: handle)?: team-foreground/i); + assert.match(manifest, /round(?: id)?: round-3/i); + assert.match(manifest, /round(?: index)?: 3/i); + assert.match(manifest, /objective: third objective/i); + assert.match(manifest, /completed/i); + assert.match(manifest, /session-a/); + assert.match(manifest, /\/sessions\/a\.jsonl/); +}); From d5bcf19100afee4b887e291df033304621990ec1 Mon Sep 17 00:00:00 2001 From: Ying Xiang Date: Thu, 20 Aug 2026 13:03:14 +0800 Subject: [PATCH 3/3] fix(pi-agent-team): expose explicit round lifecycle --- packages/pi-agent-team/README.md | 2 +- packages/pi-agent-team/src/extension.ts | 34 ++++++++++++--- packages/pi-agent-team/src/run-manager.ts | 52 +++++++++++++++++++---- 3 files changed, 73 insertions(+), 15 deletions(-) diff --git a/packages/pi-agent-team/README.md b/packages/pi-agent-team/README.md index d3e6589..ea0435e 100644 --- a/packages/pi-agent-team/README.md +++ b/packages/pi-agent-team/README.md @@ -22,7 +22,7 @@ pi -e npm:@geminixiang/pi-agent-team ## Lifecycle API migration -Existing callers do not need to change the handle they store: the tool parameter remains named `runId` for compatibility, but its value now identifies the retained team (`teamId`), not an individual execution. Use snapshot `roundId`/`roundIndex` to correlate one objective execution. `status` remains the latest round status; use `lifecycle` to distinguish an `available`, `running`, `closing`, or `closed` retained team. Terminal `rounds` are bounded to the latest 16 summaries, and restricted message bodies are never included. +Existing callers do not need to change the handle they store: the tool parameter remains named `runId` for compatibility, but its value now identifies the retained team (`teamId`), not an individual execution. Snapshots expose team lifecycle through both `teamStatus` and its `lifecycle` alias, while `latestRound`/`currentRound` explicitly identify the current or latest `RoundRun`; the existing flat `roundId`, `roundIndex`, objective, result, and round `status` fields remain compatibility aliases. Terminal `rounds` are bounded to the latest 16 summaries, and restricted message bodies are never included. `team_cancel` is idempotent for the latest cancelled round. Callers that retry asynchronously may pass the observed optional `roundId`; if a newer round has started, the stale cancellation is a no-op. Once cancellation settles, `team_prompt` starts a fresh round on the same member `sessionId`/`sessionRef` values. Events and late runtime callbacks are round-scoped so an older completion cannot settle or modify a newer round. diff --git a/packages/pi-agent-team/src/extension.ts b/packages/pi-agent-team/src/extension.ts index e938106..fbb90d4 100644 --- a/packages/pi-agent-team/src/extension.ts +++ b/packages/pi-agent-team/src/extension.ts @@ -207,6 +207,7 @@ export default function agentTeam(pi: ExtensionAPI): void { text: renderFinalContent(result, params.members, { roundId: retainedSnapshot.roundId, roundIndex: retainedSnapshot.roundIndex, + objective: retainedSnapshot.objective, }), }], details: finalDetails(details(), result), @@ -340,7 +341,7 @@ function registerTeamCancel(pi: ExtensionAPI, runs: TeamRunManager): void { function renderRunSnapshot(snapshot: TeamRunSnapshot): string { const lines = [ `team: ${snapshot.teamId}`, - `lifecycle: ${snapshot.lifecycle}`, + `team lifecycle: ${snapshot.lifecycle}`, `round: ${snapshot.roundId} (#${snapshot.roundIndex})`, `objective: ${snapshot.objective}`, `round status: ${snapshot.status}`, @@ -408,8 +409,21 @@ function summarizeLive(details: TeamDisplayDetails): string { export function renderFinalContent( result: TeamResult, members: readonly { id: string; name: string }[], - round?: { roundId: string; roundIndex: number }, + round?: { roundId: string; roundIndex: number; objective?: string }, ): string { + const resultRound = result as TeamResult & { + roundId?: string; + roundIndex?: number; + objective?: string; + }; + const manifestRound = round ?? + (resultRound.roundId !== undefined && resultRound.roundIndex !== undefined + ? { + roundId: resultRound.roundId, + roundIndex: resultRound.roundIndex, + objective: resultRound.objective, + } + : undefined); const nameOf = (id: string) => members.find((member) => member.id === id)?.name ?? id; const lines: string[] = []; if (result.report) { @@ -429,9 +443,19 @@ export function renderFinalContent( } lines.push( "TEAM MANIFEST", - `team: ${result.teamId}`, - ...(round ? [`round: ${round.roundId} (#${round.roundIndex})`] : []), - `settlement: ${result.settlement.kind} (${result.settlement.meaning}; objective correctness unverified)`, + `team handle: ${result.teamId}`, + ...(manifestRound + ? [ + `round id: ${manifestRound.roundId}`, + `round index: ${manifestRound.roundIndex}`, + ...(manifestRound.objective !== undefined + ? [`objective: ${manifestRound.objective}`] + : []), + `round outcome: ${result.settlement.kind} (${result.settlement.meaning}; objective correctness unverified)`, + ] + : [ + `settlement: ${result.settlement.kind} (${result.settlement.meaning}; objective correctness unverified)`, + ]), "members:", ); for (const member of result.members) { diff --git a/packages/pi-agent-team/src/run-manager.ts b/packages/pi-agent-team/src/run-manager.ts index 5e5d1b7..5a03045 100644 --- a/packages/pi-agent-team/src/run-manager.ts +++ b/packages/pi-agent-team/src/run-manager.ts @@ -54,7 +54,7 @@ export interface TeamRoundSummary { roundId: string; roundIndex: number; objective: string; - status: "settled" | "cancelled" | "failed"; + status: TeamRunStatus; startedAt: number; updatedAt: number; cancellation?: { requested: true; requestedAt: number; reason: string }; @@ -70,6 +70,12 @@ export interface TeamRunSnapshot { runId: string; teamId: string; lifecycle: TeamLifecycleStatus; + /** Explicit team-level lifecycle alias for operator-facing consumers. */ + teamStatus: TeamLifecycleStatus; + /** Explicit current/latest RoundRun; flat fields below remain compatibility aliases. */ + latestRound: TeamRoundSummary; + currentRound: TeamRoundSummary; + members: TeamRunResultSummary["members"]; roundId: string; roundIndex: number; objective: string; @@ -209,7 +215,7 @@ export class TeamRunManager { record.status = "running"; record.roundId = nextRoundId; record.roundIndex += 1; - record.objective = `Requester continuation directed to member ${memberId}`; + record.objective = message; record.generation += 1; record.startedAt = Date.now(); record.updatedAt = record.startedAt; @@ -233,6 +239,7 @@ export class TeamRunManager { record.cancellation = Object.freeze({ requested: true, requestedAt: Date.now(), reason: truncateUtf8(reason, MAX_EVENT_SUMMARY_BYTES) }); this.bump(record, "cancel-requested", record.cancellation.reason); record.controller.abort(new Error(record.cancellation.reason)); + queueMicrotask(() => this.finalizeCancellation(record, record.generation)); return this.snapshot(record); } @@ -292,10 +299,7 @@ export class TeamRunManager { (result) => { if (record.generation !== generation || record.runtime !== runtime) return; if (controller.signal.aborted) { - record.status = "cancelled"; - record.error = truncateUtf8(errorMessage(controller.signal.reason), MAX_EVENT_SUMMARY_BYTES); - this.bump(record, "cancelled", record.error); - this.archiveRound(record); + this.finalizeCancellation(record, generation); return; } record.status = "settled"; @@ -305,22 +309,35 @@ export class TeamRunManager { }, (cause) => { if (record.generation !== generation || record.runtime !== runtime) return; + if (controller.signal.aborted) { + this.finalizeCancellation(record, generation, cause); + return; + } const message = truncateUtf8(errorMessage(cause), MAX_EVENT_SUMMARY_BYTES); record.error = message; - record.status = controller.signal.aborted ? "cancelled" : "failed"; - this.bump(record, record.status, message); + record.status = "failed"; + this.bump(record, "failed", message); this.archiveRound(record); }, ); } + private finalizeCancellation(record: TeamRecord, generation: number, cause?: unknown): void { + if (record.generation !== generation || record.status !== "cancelling") return; + const cancellationCause = cause ?? record.controller.signal.reason ?? new Error("cancelled"); + record.error = truncateUtf8(errorMessage(cancellationCause), MAX_EVENT_SUMMARY_BYTES); + record.status = "cancelled"; + this.bump(record, "cancelled", record.error); + this.archiveRound(record); + } + private archiveRound(record: TeamRecord): void { const summary: TeamRoundSummary = Object.freeze({ teamId: record.teamId, roundId: record.roundId, roundIndex: record.roundIndex, objective: truncateUtf8(record.objective, MAX_OBJECTIVE_BYTES), - status: record.status as TeamRoundSummary["status"], + status: record.status, startedAt: record.startedAt, updatedAt: record.updatedAt, cancellation: record.cancellation, @@ -360,10 +377,27 @@ export class TeamRunManager { private snapshot(record: TeamRecord): TeamRunSnapshot { const lifecycle: TeamLifecycleStatus = record.status === "running" ? "running" : record.status === "cancelling" ? "closing" : "available"; + const latestRound: TeamRoundSummary = Object.freeze({ + teamId: record.teamId, + roundId: record.roundId, + roundIndex: record.roundIndex, + objective: truncateUtf8(record.objective, MAX_OBJECTIVE_BYTES), + status: record.status, + startedAt: record.startedAt, + updatedAt: record.updatedAt, + cancellation: record.cancellation, + result: record.result, + error: record.error, + }); + const retainedMembers = record.result?.members ?? [...record.rounds].reverse().find((round) => round.result)?.result?.members ?? Object.freeze([]); return Object.freeze({ runId: record.teamId, teamId: record.teamId, lifecycle, + teamStatus: lifecycle, + latestRound, + currentRound: latestRound, + members: retainedMembers, roundId: record.roundId, roundIndex: record.roundIndex, objective: truncateUtf8(record.objective, MAX_OBJECTIVE_BYTES),