feat(compound-engineering): live agent output, steering, and a real Q&A surface
Users can now watch everything the agent does while a CE stage works, steer
it mid-stage, and read the whole conversation as a proper chat surface.
Live output:
- New host capability: CreateInteractiveAiSessionOptions.onProgress — the
engine adapter streams thinking/text deltas + tool start/end markers from
the pi agent hooks (any plugin can use this).
- Orchestrator buffers per-session live activity (merged deltas, discrete
tool lines, capped), emits throttled progress events over SSE, and
GET /sessions/:id attaches it as liveActivity for the polling fallback.
- Routes detach turn execution: start/answer/resume return immediately
(status active) and clients converge via push/poll — the turn is watchable
instead of hidden inside a blocking POST.
- Turn timeout is now INACTIVITY-based: an actively-working long turn is
never killed; a quiet one interrupts with its working trace preserved.
- On settle the trace persists into history as a condensed record.
Steering:
- Stage protocol: responses may be a direct answer, {value, comment}
(answer + guidance), or {feedback} (guidance without answering); the
system prompt instructs agents to treat steering as first-class input.
- CeFlow: guidance textarea alongside selectable questions — attach to the
clicked answer, or "Send guidance" on its own.
Q&A UI:
- Transcript no longer hides control records: past questions/answers render
as chat bubbles (option ids → labels), steering turns marked, working
traces as collapsible "Agent work" blocks, completion marker.
- Live working pane (pulse + streaming thinking/tool lines) while a turn runs.
Tests: 130 plugin tests green (14 new: live buffer/flush ordering, inactivity
watchdog survives active work, detached convergence, steering payload shapes,
transcript rendering, live pane). Engine seam tests green; plugin/core/
engine/dashboard tsc clean. Core full suite OOMs locally (known orchestrator-
shell issue) — covered by CI shards.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
@@ -78,18 +78,44 @@ activity; from there you can:
|
||||
The list refreshes on any CE push event and falls back to polling
|
||||
`GET /sessions` while any session has a turn in flight.
|
||||
|
||||
### Live working output, steering, and the Q&A surface
|
||||
|
||||
Turn execution is **detached**: `POST /sessions`, `/answer`, and `/resume`
|
||||
return as soon as the session row reflects the request, with the agent turn
|
||||
running in the background. While it runs:
|
||||
|
||||
- The engine streams **live progress** through the seam's `onProgress` option
|
||||
(thinking/text deltas + tool start/end markers — a host capability any
|
||||
plugin can use). The orchestrator accumulates it per session and
|
||||
`GET /sessions/:id` attaches it as `liveActivity`, so the flow renders a
|
||||
live working pane (pulsing indicator, muted thinking, per-tool ✓/✗ lines).
|
||||
- The per-turn timeout is **inactivity-based**: a long but actively-working
|
||||
turn is never killed; only a turn with no progress for `turnIntervalMs` is
|
||||
interrupted (its working trace is preserved in the transcript).
|
||||
- On settle, the working trace is persisted into the conversation history as a
|
||||
condensed collapsible "Agent work" block — the transcript keeps the full
|
||||
story: opening message, every past question and answer (option ids rendered
|
||||
as labels), steering turns, working traces, and completion.
|
||||
|
||||
**Steering**: alongside any selectable question the user can type free-text
|
||||
guidance — attached to their answer as `{value, comment}`, or sent WITHOUT
|
||||
answering as `{feedback}`. The stage system prompt instructs the agent to
|
||||
treat both as first-class input (incorporate, adjust course, re-ask or
|
||||
proceed).
|
||||
|
||||
### Transport
|
||||
|
||||
Session updates are **pushed** over the shared `/api/events` SSE stream. The
|
||||
orchestrator emits observable events via `ctx.emitEvent` (turn / question /
|
||||
completed / error / interrupted); the host forwards them to connected clients as
|
||||
project-scoped `plugin:custom` events, and the view subscribes through the
|
||||
`subscribePluginEvents` context capability — refetching the session on each event
|
||||
(no raw `EventSource`; no deep dashboard import). Client **polling of
|
||||
`GET /sessions/:id` remains as a fallback** while a turn is mid-flight, so a
|
||||
missed event still converges. Session identity is project-scoped: the `projectId`
|
||||
used at `start` is threaded through every later answer/resume/poll so they
|
||||
resolve the same store and live handle.
|
||||
completed / error / interrupted, plus throttled mid-turn progress); the host
|
||||
forwards them to connected clients as project-scoped `plugin:custom` events,
|
||||
and the view subscribes through the `subscribePluginEvents` context capability
|
||||
— refetching the session on each event (no raw `EventSource`; no deep
|
||||
dashboard import). Client **polling of `GET /sessions/:id` remains as a
|
||||
fallback** while a turn is mid-flight, so a missed event still converges.
|
||||
Session identity is project-scoped: the `projectId` used at `start` is
|
||||
threaded through every later answer/resume/poll so they resolve the same store
|
||||
and live handle.
|
||||
|
||||
## Work → board bridge
|
||||
|
||||
|
||||
@@ -0,0 +1,190 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type {
|
||||
CreateInteractiveAiSessionFactory,
|
||||
InteractiveAiSessionEvent,
|
||||
InteractiveAiSessionProgressEvent,
|
||||
PlanningQuestion,
|
||||
} from "@fusion/core";
|
||||
import { buildStageSystemPrompt, CeOrchestrator, CE_EVENTS } from "../session/orchestrator.js";
|
||||
import { getStage } from "../session/stage-registry.js";
|
||||
import { makeHarness, makeScriptedSession, type TestHarness } from "./_harness.js";
|
||||
|
||||
/**
|
||||
* Live working-output + steering-protocol coverage:
|
||||
* - mid-turn progress (thinking/text deltas, tool markers) is visible via
|
||||
* getLiveActivity while the turn runs, emitted as observable events, and
|
||||
* persisted into history as a condensed trace when the turn settles;
|
||||
* - the turn timeout is INACTIVITY-based — an actively-working long turn is
|
||||
* never killed, a quiet one is interrupted with its trace preserved;
|
||||
* - detached start/answer return immediately and converge via persisted state;
|
||||
* - the stage system prompt documents the steering response shapes.
|
||||
*/
|
||||
|
||||
const QUESTION: PlanningQuestion = { id: "q1", type: "text", question: "Topic?" };
|
||||
|
||||
let h: TestHarness;
|
||||
beforeEach(() => {
|
||||
h = makeHarness();
|
||||
});
|
||||
afterEach(() => {
|
||||
h.close();
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
function deferred<T>() {
|
||||
let resolve!: (v: T) => void;
|
||||
const promise = new Promise<T>((r) => {
|
||||
resolve = r;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
|
||||
|
||||
/** A factory exposing the onProgress hook and a controllable nextEvent. */
|
||||
function progressFactory(nextEvent: () => Promise<InteractiveAiSessionEvent>) {
|
||||
const captured: { progress?: (e: InteractiveAiSessionProgressEvent) => void; dispose: ReturnType<typeof vi.fn> } = {
|
||||
dispose: vi.fn(),
|
||||
};
|
||||
const factory: CreateInteractiveAiSessionFactory = vi.fn(async (opts) => {
|
||||
captured.progress = opts.onProgress;
|
||||
return {
|
||||
session: {
|
||||
prompt: vi.fn(async () => undefined),
|
||||
answer: vi.fn(async () => undefined),
|
||||
nextEvent,
|
||||
dispose: captured.dispose,
|
||||
},
|
||||
};
|
||||
});
|
||||
return { factory, captured };
|
||||
}
|
||||
|
||||
describe("live working output", () => {
|
||||
it("buffers mid-turn progress, emits observable events, and persists the trace on settle (before the question)", async () => {
|
||||
const evt = deferred<InteractiveAiSessionEvent>();
|
||||
const { factory, captured } = progressFactory(() => evt.promise);
|
||||
const orch = new CeOrchestrator({
|
||||
ctx: h.ctx,
|
||||
createInteractiveAiSession: factory,
|
||||
projectRoot: h.projectRoot,
|
||||
turnTimeoutMs: 5000,
|
||||
});
|
||||
|
||||
const started = await orch.start("brainstorm", { openingMessage: "go", detach: true });
|
||||
expect(["launching", "active"]).toContain(started.session.status);
|
||||
await vi.waitFor(() => expect(captured.progress).toBeDefined());
|
||||
|
||||
// Stream: consecutive deltas of one kind merge; tool start/end are discrete.
|
||||
captured.progress!({ type: "thinking", delta: "Let me " });
|
||||
captured.progress!({ type: "thinking", delta: "look around." });
|
||||
captured.progress!({ type: "tool", name: "Read", phase: "start" });
|
||||
captured.progress!({ type: "tool", name: "Read", phase: "end", isError: false });
|
||||
captured.progress!({ type: "text", delta: "Drafting…" });
|
||||
|
||||
const live = orch.getLiveActivity(started.session.id);
|
||||
expect(live.map((t) => t.kind)).toEqual(["thinking", "tool", "text"]);
|
||||
expect(live[0].text).toBe("Let me look around.");
|
||||
expect(live[1].done).toBe(true);
|
||||
expect(live[1].isError).toBeUndefined();
|
||||
|
||||
// Observable progress event emitted (throttled; the first one is immediate).
|
||||
expect(
|
||||
h.emitted.some((e) => e.event === CE_EVENTS.turn && (e.data as { kind?: string }).kind === "progress"),
|
||||
).toBe(true);
|
||||
|
||||
// Settle the turn → buffer flushed into history BEFORE the question record.
|
||||
evt.resolve({ type: "question", data: QUESTION });
|
||||
await vi.waitFor(() => expect(orch.getState(started.session.id)?.status).toBe("awaiting_input"));
|
||||
expect(orch.getLiveActivity(started.session.id)).toHaveLength(0);
|
||||
|
||||
const history = orch.getState(started.session.id)!.conversationHistory;
|
||||
const activityIdx = history.findIndex((t) => t.role === "agent" && t.text.startsWith('{"activity"'));
|
||||
const questionIdx = history.findIndex((t) => t.role === "agent" && t.text.startsWith('{"question"'));
|
||||
expect(activityIdx).toBeGreaterThanOrEqual(0);
|
||||
expect(questionIdx).toBeGreaterThan(activityIdx);
|
||||
const trace = JSON.parse(history[activityIdx].text) as {
|
||||
activity: { turns: Array<{ kind: string; text: string }> };
|
||||
};
|
||||
expect(trace.activity.turns.map((t) => t.kind)).toEqual(["thinking", "tool", "text"]);
|
||||
});
|
||||
|
||||
it("inactivity watchdog: an actively-working long turn survives past the timeout; a quiet one is interrupted with its trace kept", async () => {
|
||||
const { factory, captured } = progressFactory(() => new Promise<InteractiveAiSessionEvent>(() => undefined));
|
||||
const orch = new CeOrchestrator({
|
||||
ctx: h.ctx,
|
||||
createInteractiveAiSession: factory,
|
||||
projectRoot: h.projectRoot,
|
||||
turnTimeoutMs: 120,
|
||||
});
|
||||
const started = await orch.start("brainstorm", { openingMessage: "go", detach: true });
|
||||
const id = started.session.id;
|
||||
await vi.waitFor(() => expect(captured.progress).toBeDefined());
|
||||
|
||||
// Keep working for ~3× the timeout — must NOT be interrupted.
|
||||
for (let i = 0; i < 8; i++) {
|
||||
await sleep(45);
|
||||
captured.progress!({ type: "thinking", delta: "." });
|
||||
}
|
||||
expect(orch.getState(id)?.status).toBe("active");
|
||||
|
||||
// Go quiet → interrupted after the inactivity window, trace preserved.
|
||||
await vi.waitFor(() => expect(orch.getState(id)?.status).toBe("interrupted"), { timeout: 2000 });
|
||||
expect(orch.getState(id)?.error).toMatch(/no agent activity/i);
|
||||
const history = orch.getState(id)!.conversationHistory;
|
||||
expect(history.some((t) => t.text.startsWith('{"activity"'))).toBe(true);
|
||||
expect(captured.dispose).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("detached turns (route posture)", () => {
|
||||
it("answer(detach) returns immediately with status active and converges to the next question", async () => {
|
||||
const NEXT: PlanningQuestion = { id: "q2", type: "text", question: "More?" };
|
||||
const scripted = makeScriptedSession([
|
||||
{ type: "question", data: QUESTION },
|
||||
{ type: "question", data: NEXT },
|
||||
]);
|
||||
const orch = new CeOrchestrator({
|
||||
ctx: h.ctx,
|
||||
createInteractiveAiSession: vi.fn(async () => ({ session: scripted })),
|
||||
projectRoot: h.projectRoot,
|
||||
turnTimeoutMs: 5000,
|
||||
});
|
||||
const started = await orch.start("brainstorm", { openingMessage: "go" });
|
||||
expect(started.session.status).toBe("awaiting_input");
|
||||
|
||||
const stepped = await orch.answer(started.session.id, "q1", "widgets", { detach: true });
|
||||
// Detached return reflects the just-accepted answer, not the settled turn…
|
||||
expect(stepped.session.status).toBe("active");
|
||||
expect(stepped.session.currentQuestion).toBeNull();
|
||||
// …and the background turn converges to the next question.
|
||||
await vi.waitFor(() => expect(orch.getState(started.session.id)?.currentQuestion?.id).toBe("q2"));
|
||||
expect(orch.getState(started.session.id)?.status).toBe("awaiting_input");
|
||||
});
|
||||
|
||||
it("start(detach) without a working factory converges to an error state (never silent)", async () => {
|
||||
const orch = new CeOrchestrator({
|
||||
ctx: h.ctx,
|
||||
createInteractiveAiSession: vi.fn(async () => {
|
||||
throw new Error("factory exploded");
|
||||
}),
|
||||
projectRoot: h.projectRoot,
|
||||
turnTimeoutMs: 5000,
|
||||
});
|
||||
const started = await orch.start("brainstorm", { openingMessage: "go", detach: true });
|
||||
expect(started.session.id).toBeTruthy();
|
||||
await vi.waitFor(() => expect(orch.getState(started.session.id)?.status).toBe("error"));
|
||||
expect(orch.getState(started.session.id)?.error).toContain("factory exploded");
|
||||
expect(h.emitted.map((e) => e.event)).toContain(CE_EVENTS.error);
|
||||
});
|
||||
});
|
||||
|
||||
describe("steering protocol", () => {
|
||||
it("the stage system prompt documents direct, value+comment, and feedback-only response shapes", () => {
|
||||
const prompt = buildStageSystemPrompt(getStage("brainstorm")!);
|
||||
expect(prompt).toContain('"value"');
|
||||
expect(prompt).toContain('"comment"');
|
||||
expect(prompt).toContain('"feedback"');
|
||||
expect(prompt).toMatch(/steering/i);
|
||||
});
|
||||
});
|
||||
@@ -1,15 +1,24 @@
|
||||
import { useMemo, useState } from "react";
|
||||
import type { PlanningQuestion } from "@fusion/core";
|
||||
import type { CeConversationTurn, CeSession } from "../session/session-store.js";
|
||||
import type { CeActivityTurn, CeConversationTurn, CeSession } from "../session/session-store.js";
|
||||
import { canRenderRichly } from "./ce-question-support.js";
|
||||
|
||||
/**
|
||||
* CeFlow — the interactive renderer (U6).
|
||||
*
|
||||
* Renders the four interaction types CeFlow expresses richly (`text`,
|
||||
* `single_select`, `multi_select`, `confirm`) plus streamed `thinking`/`text`
|
||||
* history. When a turn carries a question CeFlow CANNOT express, it degrades to
|
||||
* a plain chat view that is VISUALLY MARKED as degraded (R8/AE1) — the stage is
|
||||
* `single_select`, `multi_select`, `confirm`), the FULL conversation so far —
|
||||
* past questions and answers as proper chat bubbles, the agent's working
|
||||
* traces (thinking / tool activity) as collapsible blocks — and, while a turn
|
||||
* runs, a LIVE working pane streaming the agent's current output.
|
||||
*
|
||||
* Steering: alongside any selectable question the user can attach free-text
|
||||
* guidance to their answer (`{value, comment}`) or send guidance WITHOUT
|
||||
* answering (`{feedback}`) — the stage system prompt instructs the agent to
|
||||
* treat both as first-class input.
|
||||
*
|
||||
* When a turn carries a question CeFlow CANNOT express, it degrades to a
|
||||
* plain chat view that is VISUALLY MARKED as degraded (R8/AE1) — the stage is
|
||||
* still completable there via a free-text answer.
|
||||
*
|
||||
* It does NOT import `PlanningModeModal` or any dashboard internal (KTD3 scope
|
||||
@@ -28,28 +37,181 @@ export interface CeFlowProps {
|
||||
onClose?: () => void;
|
||||
}
|
||||
|
||||
/** Render the agent/user conversation so far (streamed thinking/text). */
|
||||
// ── Transcript parsing ───────────────────────────────────────────────────────
|
||||
|
||||
type DisplayItem =
|
||||
| { kind: "chat"; role: "user" | "agent"; text: string }
|
||||
| { kind: "qa-question"; question: PlanningQuestion }
|
||||
| { kind: "qa-answer"; question?: PlanningQuestion; response: unknown }
|
||||
| { kind: "activity"; turns: CeActivityTurn[] }
|
||||
| { kind: "complete" };
|
||||
|
||||
function tryParseJson(text: string): Record<string, unknown> | undefined {
|
||||
if (!text.startsWith("{")) return undefined;
|
||||
try {
|
||||
const parsed: unknown = JSON.parse(text);
|
||||
return parsed && typeof parsed === "object" && !Array.isArray(parsed)
|
||||
? (parsed as Record<string, unknown>)
|
||||
: undefined;
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Turn the persisted history (chat turns + serialized control records) into
|
||||
* renderable items. Control records are no longer hidden — questions, answers,
|
||||
* and working traces are the conversation.
|
||||
*/
|
||||
function parseHistory(history: CeConversationTurn[]): DisplayItem[] {
|
||||
const items: DisplayItem[] = [];
|
||||
const questionsById = new Map<string, PlanningQuestion>();
|
||||
for (const turn of history) {
|
||||
const obj = tryParseJson(turn.text);
|
||||
if (obj && turn.role === "agent") {
|
||||
const q = obj.question as PlanningQuestion | undefined;
|
||||
if (q && typeof q.id === "string" && typeof q.question === "string") {
|
||||
questionsById.set(q.id, q);
|
||||
items.push({ kind: "qa-question", question: q });
|
||||
continue;
|
||||
}
|
||||
const activity = obj.activity as { turns?: CeActivityTurn[] } | undefined;
|
||||
if (activity && Array.isArray(activity.turns)) {
|
||||
items.push({ kind: "activity", turns: activity.turns });
|
||||
continue;
|
||||
}
|
||||
if (obj.complete === true) {
|
||||
items.push({ kind: "complete" });
|
||||
continue;
|
||||
}
|
||||
}
|
||||
if (obj && turn.role === "user" && "answer" in obj) {
|
||||
items.push({
|
||||
kind: "qa-answer",
|
||||
question: typeof obj.questionId === "string" ? questionsById.get(obj.questionId) : undefined,
|
||||
response: obj.answer,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
items.push({ kind: "chat", role: turn.role, text: turn.text });
|
||||
}
|
||||
return items;
|
||||
}
|
||||
|
||||
/** Human-readable rendering of an answer payload (option ids → labels). */
|
||||
function formatAnswer(
|
||||
response: unknown,
|
||||
question?: PlanningQuestion,
|
||||
): { main: string; comment?: string; feedbackOnly?: boolean } {
|
||||
if (response && typeof response === "object" && !Array.isArray(response)) {
|
||||
const r = response as Record<string, unknown>;
|
||||
if (typeof r.feedback === "string") return { main: r.feedback, feedbackOnly: true };
|
||||
if ("value" in r) {
|
||||
const base = formatAnswer(r.value, question);
|
||||
return {
|
||||
main: base.main,
|
||||
...(typeof r.comment === "string" && r.comment ? { comment: r.comment } : {}),
|
||||
};
|
||||
}
|
||||
}
|
||||
const label = (id: unknown): string =>
|
||||
question?.options?.find((o) => o.id === id)?.label ?? String(id);
|
||||
if (Array.isArray(response)) return { main: response.map(label).join(", ") };
|
||||
if (typeof response === "boolean") return { main: response ? "Yes" : "No" };
|
||||
return { main: label(response) };
|
||||
}
|
||||
|
||||
// ── Working-trace rendering ──────────────────────────────────────────────────
|
||||
|
||||
/** Render thinking/text/tool activity turns (persisted trace or live pane). */
|
||||
function ActivityTrace({ turns, live }: { turns: CeActivityTurn[]; live?: boolean }) {
|
||||
return (
|
||||
<div
|
||||
className={`ce-flow-activity${live ? " is-live" : ""}`}
|
||||
data-testid={live ? "ce-flow-live-activity" : "ce-flow-activity-trace"}
|
||||
>
|
||||
{turns.map((t, i) =>
|
||||
t.kind === "tool" ? (
|
||||
<div
|
||||
key={i}
|
||||
className={`ce-activity-tool${t.isError ? " is-error" : t.done ? " is-done" : " is-running"}`}
|
||||
data-testid="ce-activity-tool"
|
||||
>
|
||||
<span className="ce-activity-tool-marker">{t.isError ? "✗" : t.done ? "✓" : "▸"}</span> {t.text}
|
||||
</div>
|
||||
) : (
|
||||
<pre key={i} className={`ce-activity-block ce-activity-${t.kind}`} data-kind={t.kind}>
|
||||
{t.text}
|
||||
</pre>
|
||||
),
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
/** Render the full conversation: chat, Q&A bubbles, and working traces. */
|
||||
function Transcript({ history }: { history: CeConversationTurn[] }) {
|
||||
const visible = history.filter((t) => {
|
||||
// Hide serialized question/answer/complete markers from the readable
|
||||
// transcript; they are control records, not chat.
|
||||
if (t.role === "agent" && /^\{"(question|complete)"/.test(t.text)) return false;
|
||||
if (t.role === "user" && /^\{"answer"/.test(t.text)) return false;
|
||||
return true;
|
||||
});
|
||||
if (visible.length === 0) return null;
|
||||
const items = useMemo(() => parseHistory(history), [history]);
|
||||
if (items.length === 0) return null;
|
||||
return (
|
||||
<ol className="ce-flow-transcript" data-testid="ce-flow-transcript">
|
||||
{visible.map((turn, i) => (
|
||||
<li key={i} className={`ce-flow-turn ce-flow-turn-${turn.role}`} data-role={turn.role}>
|
||||
<span className="ce-flow-turn-role">{turn.role === "agent" ? "Agent" : "You"}</span>
|
||||
<span className="ce-flow-turn-text">{turn.text}</span>
|
||||
</li>
|
||||
))}
|
||||
{items.map((item, i) => {
|
||||
switch (item.kind) {
|
||||
case "chat":
|
||||
return (
|
||||
<li key={i} className={`ce-flow-turn ce-flow-turn-${item.role}`} data-role={item.role}>
|
||||
<span className="ce-flow-turn-role">{item.role === "agent" ? "Agent" : "You"}</span>
|
||||
<span className="ce-flow-turn-text">{item.text}</span>
|
||||
</li>
|
||||
);
|
||||
case "qa-question":
|
||||
return (
|
||||
<li key={i} className="ce-flow-turn ce-flow-turn-agent ce-flow-turn-question" data-testid="ce-flow-past-question">
|
||||
<span className="ce-flow-turn-role">Agent asked</span>
|
||||
<span className="ce-flow-turn-text">{item.question.question}</span>
|
||||
</li>
|
||||
);
|
||||
case "qa-answer": {
|
||||
const a = formatAnswer(item.response, item.question);
|
||||
return (
|
||||
<li
|
||||
key={i}
|
||||
className={`ce-flow-turn ce-flow-turn-user ce-flow-turn-answer${a.feedbackOnly ? " is-steering" : ""}`}
|
||||
data-testid="ce-flow-past-answer"
|
||||
>
|
||||
<span className="ce-flow-turn-role">{a.feedbackOnly ? "You steered" : "You answered"}</span>
|
||||
<span className="ce-flow-turn-text">{a.main}</span>
|
||||
{a.comment ? (
|
||||
<span className="ce-flow-turn-comment" data-testid="ce-flow-answer-comment">
|
||||
{a.comment}
|
||||
</span>
|
||||
) : null}
|
||||
</li>
|
||||
);
|
||||
}
|
||||
case "activity":
|
||||
return (
|
||||
<li key={i} className="ce-flow-turn ce-flow-turn-agent ce-flow-turn-activity">
|
||||
<details className="ce-flow-activity-details" data-testid="ce-flow-activity">
|
||||
<summary>Agent work ({item.turns.length} step{item.turns.length === 1 ? "" : "s"})</summary>
|
||||
<ActivityTrace turns={item.turns} />
|
||||
</details>
|
||||
</li>
|
||||
);
|
||||
case "complete":
|
||||
return (
|
||||
<li key={i} className="ce-flow-turn ce-flow-turn-agent ce-flow-turn-done">
|
||||
<span className="ce-flow-turn-text">✓ Stage complete</span>
|
||||
</li>
|
||||
);
|
||||
}
|
||||
})}
|
||||
</ol>
|
||||
);
|
||||
}
|
||||
|
||||
// ── Question rendering ───────────────────────────────────────────────────────
|
||||
|
||||
/** Rich renderer for a single supported question type. */
|
||||
function RichQuestion({
|
||||
question,
|
||||
@@ -228,11 +390,85 @@ function DegradedQuestion({
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Question panel with steering. Wraps the rich/degraded renderer and adds the
|
||||
* guidance channel for selectable questions:
|
||||
* - guidance typed + an option clicked → `{value, comment}` (answer + steer),
|
||||
* - guidance typed + "Send guidance" → `{feedback}` (steer without answering).
|
||||
* Free-text questions skip the extra box — their answer field already takes
|
||||
* the user's own words.
|
||||
*/
|
||||
function QuestionPanel({
|
||||
question,
|
||||
disabled,
|
||||
onAnswer,
|
||||
}: {
|
||||
question: PlanningQuestion;
|
||||
disabled: boolean;
|
||||
onAnswer: (questionId: string, response: unknown) => void;
|
||||
}) {
|
||||
const [guidance, setGuidance] = useState("");
|
||||
const rich = canRenderRichly(question);
|
||||
|
||||
const submitWithGuidance = (questionId: string, response: unknown) => {
|
||||
const comment = guidance.trim();
|
||||
onAnswer(questionId, comment ? { value: response, comment } : response);
|
||||
setGuidance("");
|
||||
};
|
||||
|
||||
const sendGuidanceOnly = () => {
|
||||
const feedback = guidance.trim();
|
||||
if (!feedback) return;
|
||||
onAnswer(question.id, { feedback });
|
||||
setGuidance("");
|
||||
};
|
||||
|
||||
const showGuidance = rich && question.type !== "text";
|
||||
|
||||
return (
|
||||
<div className="ce-flow-question-panel">
|
||||
{rich ? (
|
||||
<RichQuestion question={question} disabled={disabled} onAnswer={submitWithGuidance} />
|
||||
) : (
|
||||
<DegradedQuestion question={question} disabled={disabled} onAnswer={onAnswer} />
|
||||
)}
|
||||
{showGuidance ? (
|
||||
<div className="ce-flow-guidance" data-testid="ce-flow-guidance">
|
||||
<label className="ce-flow-guidance-label" htmlFor="ce-flow-guidance-input">
|
||||
Steer in your own words (optional — attached to your answer, or sent on its own)
|
||||
</label>
|
||||
<div className="ce-flow-guidance-row">
|
||||
<textarea
|
||||
id="ce-flow-guidance-input"
|
||||
data-testid="ce-flow-guidance-input"
|
||||
value={guidance}
|
||||
disabled={disabled}
|
||||
onChange={(e) => setGuidance(e.target.value)}
|
||||
rows={2}
|
||||
placeholder="e.g. focus on the mobile flow, skip auth for now…"
|
||||
/>
|
||||
<button
|
||||
type="button"
|
||||
className="btn"
|
||||
data-testid="ce-flow-guidance-send"
|
||||
disabled={disabled || !guidance.trim()}
|
||||
onClick={sendGuidanceOnly}
|
||||
>
|
||||
Send guidance
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
) : null}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
// ── Flow surface ─────────────────────────────────────────────────────────────
|
||||
|
||||
export function CeFlow(props: CeFlowProps) {
|
||||
const { session, busy, error, onAnswer, onResume, onClose } = props;
|
||||
|
||||
const question = session?.currentQuestion ?? undefined;
|
||||
const rich = useMemo(() => (question ? canRenderRichly(question) : false), [question]);
|
||||
|
||||
if (!session) {
|
||||
return (
|
||||
@@ -250,6 +486,7 @@ export function CeFlow(props: CeFlowProps) {
|
||||
const status = session.status;
|
||||
const settledTerminal = status === "completed";
|
||||
const recoverable = status === "interrupted" || status === "error";
|
||||
const working = status === "active" || status === "launching";
|
||||
|
||||
return (
|
||||
<div className="ce-flow card" data-testid="ce-flow" data-status={status} data-stage={session.stage}>
|
||||
@@ -267,10 +504,16 @@ export function CeFlow(props: CeFlowProps) {
|
||||
|
||||
<Transcript history={session.conversationHistory} />
|
||||
|
||||
{busy && status !== "awaiting_input" ? (
|
||||
<p className="ce-flow-thinking" data-testid="ce-flow-thinking">
|
||||
Thinking…
|
||||
</p>
|
||||
{working || (busy && status !== "awaiting_input") ? (
|
||||
<div className="ce-flow-working" data-testid="ce-flow-thinking">
|
||||
<p className="ce-flow-working-label">
|
||||
<span className="ce-flow-pulse" aria-hidden="true" />
|
||||
Agent working…
|
||||
</p>
|
||||
{session.liveActivity && session.liveActivity.length > 0 ? (
|
||||
<ActivityTrace turns={session.liveActivity} live />
|
||||
) : null}
|
||||
</div>
|
||||
) : null}
|
||||
|
||||
{error ? (
|
||||
@@ -280,11 +523,7 @@ export function CeFlow(props: CeFlowProps) {
|
||||
) : null}
|
||||
|
||||
{status === "awaiting_input" && question ? (
|
||||
rich ? (
|
||||
<RichQuestion question={question} disabled={Boolean(busy)} onAnswer={onAnswer} />
|
||||
) : (
|
||||
<DegradedQuestion question={question} disabled={Boolean(busy)} onAnswer={onAnswer} />
|
||||
)
|
||||
<QuestionPanel question={question} disabled={Boolean(busy)} onAnswer={onAnswer} />
|
||||
) : null}
|
||||
|
||||
{recoverable ? (
|
||||
|
||||
@@ -368,3 +368,152 @@
|
||||
font-size: 0.72rem;
|
||||
opacity: 0.6;
|
||||
}
|
||||
|
||||
/* ── Q&A transcript bubbles ────────────────────────────────────────────── */
|
||||
.ce-flow-transcript {
|
||||
list-style: none;
|
||||
margin: 0 0 0.8rem;
|
||||
padding: 0;
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
gap: 0.45rem;
|
||||
max-height: 50vh;
|
||||
overflow-y: auto;
|
||||
}
|
||||
.ce-flow-turn {
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
gap: 0.15rem;
|
||||
max-width: 85%;
|
||||
padding: 0.45rem 0.65rem;
|
||||
border-radius: 10px;
|
||||
background: color-mix(in srgb, var(--color-border, #ddd) 30%, transparent);
|
||||
}
|
||||
.ce-flow-turn-user {
|
||||
align-self: flex-end;
|
||||
background: color-mix(in srgb, var(--color-accent, #36c) 12%, transparent);
|
||||
}
|
||||
.ce-flow-turn-role {
|
||||
font-size: 0.68rem;
|
||||
font-weight: 600;
|
||||
text-transform: uppercase;
|
||||
letter-spacing: 0.04em;
|
||||
opacity: 0.55;
|
||||
}
|
||||
.ce-flow-turn-text {
|
||||
white-space: pre-wrap;
|
||||
word-break: break-word;
|
||||
font-size: 0.86rem;
|
||||
}
|
||||
.ce-flow-turn-question {
|
||||
border-left: 3px solid var(--color-accent, #36c);
|
||||
}
|
||||
.ce-flow-turn-answer.is-steering {
|
||||
border-left: 3px solid var(--color-warning, #c80);
|
||||
}
|
||||
.ce-flow-turn-comment {
|
||||
font-size: 0.78rem;
|
||||
font-style: italic;
|
||||
opacity: 0.85;
|
||||
border-top: 1px dashed color-mix(in srgb, var(--color-border, #ddd) 60%, transparent);
|
||||
padding-top: 0.25rem;
|
||||
}
|
||||
.ce-flow-turn-done {
|
||||
align-self: center;
|
||||
background: color-mix(in srgb, var(--color-success, #2a7) 10%, transparent);
|
||||
font-size: 0.8rem;
|
||||
}
|
||||
.ce-flow-turn-activity {
|
||||
background: transparent;
|
||||
padding: 0;
|
||||
max-width: 100%;
|
||||
}
|
||||
|
||||
/* ── Agent working trace (persisted + live) ─────────────────────────────── */
|
||||
.ce-flow-activity-details summary {
|
||||
cursor: pointer;
|
||||
font-size: 0.78rem;
|
||||
opacity: 0.7;
|
||||
}
|
||||
.ce-flow-activity {
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
gap: 0.25rem;
|
||||
margin: 0.3rem 0 0;
|
||||
padding: 0.5rem 0.6rem;
|
||||
border: 1px solid color-mix(in srgb, var(--color-border, #ddd) 70%, transparent);
|
||||
border-radius: 8px;
|
||||
background: color-mix(in srgb, var(--color-border, #ddd) 12%, transparent);
|
||||
max-height: 16rem;
|
||||
overflow-y: auto;
|
||||
}
|
||||
.ce-activity-block {
|
||||
margin: 0;
|
||||
font-size: 0.76rem;
|
||||
white-space: pre-wrap;
|
||||
word-break: break-word;
|
||||
font-family: var(--font-mono, ui-monospace, monospace);
|
||||
}
|
||||
.ce-activity-thinking {
|
||||
opacity: 0.6;
|
||||
font-style: italic;
|
||||
}
|
||||
.ce-activity-tool {
|
||||
font-size: 0.76rem;
|
||||
font-family: var(--font-mono, ui-monospace, monospace);
|
||||
}
|
||||
.ce-activity-tool.is-running .ce-activity-tool-marker {
|
||||
color: var(--color-accent, #36c);
|
||||
}
|
||||
.ce-activity-tool.is-done .ce-activity-tool-marker {
|
||||
color: var(--color-success, #2a7);
|
||||
}
|
||||
.ce-activity-tool.is-error .ce-activity-tool-marker {
|
||||
color: var(--color-danger, #d23);
|
||||
}
|
||||
|
||||
/* ── Live working pane ──────────────────────────────────────────────────── */
|
||||
.ce-flow-working {
|
||||
margin: 0.4rem 0;
|
||||
}
|
||||
.ce-flow-working-label {
|
||||
display: flex;
|
||||
align-items: center;
|
||||
gap: 0.45rem;
|
||||
margin: 0 0 0.3rem;
|
||||
font-size: 0.82rem;
|
||||
opacity: 0.85;
|
||||
}
|
||||
.ce-flow-pulse {
|
||||
width: 8px;
|
||||
height: 8px;
|
||||
border-radius: 50%;
|
||||
background: var(--color-accent, #36c);
|
||||
animation: ce-pulse 1.2s ease-in-out infinite;
|
||||
}
|
||||
@keyframes ce-pulse {
|
||||
0%, 100% { opacity: 0.25; transform: scale(0.8); }
|
||||
50% { opacity: 1; transform: scale(1); }
|
||||
}
|
||||
|
||||
/* ── Steering / guidance channel ────────────────────────────────────────── */
|
||||
.ce-flow-guidance {
|
||||
margin-top: 0.5rem;
|
||||
padding-top: 0.5rem;
|
||||
border-top: 1px dashed color-mix(in srgb, var(--color-border, #ddd) 70%, transparent);
|
||||
}
|
||||
.ce-flow-guidance-label {
|
||||
display: block;
|
||||
font-size: 0.74rem;
|
||||
opacity: 0.65;
|
||||
margin-bottom: 0.3rem;
|
||||
}
|
||||
.ce-flow-guidance-row {
|
||||
display: flex;
|
||||
gap: 0.4rem;
|
||||
align-items: flex-end;
|
||||
}
|
||||
.ce-flow-guidance-row textarea {
|
||||
flex: 1;
|
||||
resize: vertical;
|
||||
}
|
||||
|
||||
@@ -124,12 +124,159 @@ describe("CeFlow — degraded fallback (AE1)", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("CeFlow — steering (guidance channel)", () => {
|
||||
const q: PlanningQuestion = {
|
||||
id: "q-steer",
|
||||
type: "single_select",
|
||||
question: "Pick a direction",
|
||||
options: [
|
||||
{ id: "a", label: "Alpha" },
|
||||
{ id: "b", label: "Beta" },
|
||||
],
|
||||
};
|
||||
|
||||
it("attaches typed guidance to the chosen answer as {value, comment}", () => {
|
||||
const onAnswer = vi.fn();
|
||||
render(<CeFlow session={makeSession({ currentQuestion: q })} onAnswer={onAnswer} />);
|
||||
fireEvent.change(screen.getByTestId("ce-flow-guidance-input"), {
|
||||
target: { value: "focus on mobile" },
|
||||
});
|
||||
fireEvent.click(screen.getByText("Beta"));
|
||||
expect(onAnswer).toHaveBeenCalledWith("q-steer", { value: "b", comment: "focus on mobile" });
|
||||
});
|
||||
|
||||
it("sends guidance WITHOUT answering as {feedback}", () => {
|
||||
const onAnswer = vi.fn();
|
||||
render(<CeFlow session={makeSession({ currentQuestion: q })} onAnswer={onAnswer} />);
|
||||
const send = screen.getByTestId("ce-flow-guidance-send");
|
||||
expect(send).toBeDisabled(); // empty guidance can't be sent
|
||||
fireEvent.change(screen.getByTestId("ce-flow-guidance-input"), {
|
||||
target: { value: "skip auth for now" },
|
||||
});
|
||||
fireEvent.click(send);
|
||||
expect(onAnswer).toHaveBeenCalledWith("q-steer", { feedback: "skip auth for now" });
|
||||
});
|
||||
|
||||
it("plain answers stay unwrapped when no guidance is typed", () => {
|
||||
const onAnswer = vi.fn();
|
||||
render(<CeFlow session={makeSession({ currentQuestion: q })} onAnswer={onAnswer} />);
|
||||
fireEvent.click(screen.getByText("Alpha"));
|
||||
expect(onAnswer).toHaveBeenCalledWith("q-steer", "a");
|
||||
});
|
||||
|
||||
it("free-text questions get no extra guidance box (their answer field already takes free text)", () => {
|
||||
const textQ: PlanningQuestion = { id: "q-text", type: "text", question: "Goal?" };
|
||||
render(<CeFlow session={makeSession({ currentQuestion: textQ })} onAnswer={vi.fn()} />);
|
||||
expect(screen.queryByTestId("ce-flow-guidance")).not.toBeInTheDocument();
|
||||
});
|
||||
});
|
||||
|
||||
describe("CeFlow — Q&A transcript rendering", () => {
|
||||
const pastQ: PlanningQuestion = {
|
||||
id: "q-past",
|
||||
type: "single_select",
|
||||
question: "Which path?",
|
||||
options: [
|
||||
{ id: "x", label: "The X path" },
|
||||
{ id: "y", label: "The Y path" },
|
||||
],
|
||||
};
|
||||
|
||||
function historyWith(answer: unknown) {
|
||||
return [
|
||||
{ role: "user" as const, text: "kick off", at: "t0" },
|
||||
{ role: "agent" as const, text: JSON.stringify({ question: pastQ }), at: "t1" },
|
||||
{ role: "user" as const, text: JSON.stringify({ answer, questionId: "q-past" }), at: "t2" },
|
||||
];
|
||||
}
|
||||
|
||||
it("renders past questions and answers as bubbles, mapping option ids to labels", () => {
|
||||
render(
|
||||
<CeFlow
|
||||
session={makeSession({ status: "active", conversationHistory: historyWith("y") })}
|
||||
onAnswer={vi.fn()}
|
||||
/>,
|
||||
);
|
||||
expect(screen.getByTestId("ce-flow-past-question")).toHaveTextContent("Which path?");
|
||||
// The answer shows the LABEL, not the raw option id.
|
||||
expect(screen.getByTestId("ce-flow-past-answer")).toHaveTextContent("The Y path");
|
||||
// The opening message renders as a plain user bubble.
|
||||
expect(screen.getByText("kick off")).toBeInTheDocument();
|
||||
});
|
||||
|
||||
it("renders {value, comment} answers with the steering comment attached", () => {
|
||||
render(
|
||||
<CeFlow
|
||||
session={makeSession({
|
||||
status: "active",
|
||||
conversationHistory: historyWith({ value: "x", comment: "but keep it small" }),
|
||||
})}
|
||||
onAnswer={vi.fn()}
|
||||
/>,
|
||||
);
|
||||
expect(screen.getByTestId("ce-flow-past-answer")).toHaveTextContent("The X path");
|
||||
expect(screen.getByTestId("ce-flow-answer-comment")).toHaveTextContent("but keep it small");
|
||||
});
|
||||
|
||||
it("renders {feedback} turns as steering, not answers", () => {
|
||||
render(
|
||||
<CeFlow
|
||||
session={makeSession({
|
||||
status: "active",
|
||||
conversationHistory: historyWith({ feedback: "go another way" }),
|
||||
})}
|
||||
onAnswer={vi.fn()}
|
||||
/>,
|
||||
);
|
||||
const turn = screen.getByTestId("ce-flow-past-answer");
|
||||
expect(turn).toHaveTextContent("You steered");
|
||||
expect(turn).toHaveTextContent("go another way");
|
||||
});
|
||||
|
||||
it("renders persisted working traces as a collapsible activity block", () => {
|
||||
const history = [
|
||||
{
|
||||
role: "agent" as const,
|
||||
text: JSON.stringify({
|
||||
activity: {
|
||||
turns: [
|
||||
{ kind: "thinking", text: "Scanning the repo…", at: "t" },
|
||||
{ kind: "tool", text: "Read", at: "t", done: true },
|
||||
],
|
||||
},
|
||||
}),
|
||||
at: "t1",
|
||||
},
|
||||
];
|
||||
render(<CeFlow session={makeSession({ status: "active", conversationHistory: history })} onAnswer={vi.fn()} />);
|
||||
const details = screen.getByTestId("ce-flow-activity");
|
||||
expect(details).toHaveTextContent("Agent work (2 steps)");
|
||||
expect(screen.getByText("Scanning the repo…")).toBeInTheDocument();
|
||||
expect(screen.getByTestId("ce-activity-tool")).toHaveTextContent("Read");
|
||||
});
|
||||
});
|
||||
|
||||
describe("CeFlow — lifecycle surfaces", () => {
|
||||
it("shows thinking while a turn runs", () => {
|
||||
it("shows the working pane while a turn runs", () => {
|
||||
render(<CeFlow session={makeSession({ status: "active", currentQuestion: null })} busy onAnswer={vi.fn()} />);
|
||||
expect(screen.getByTestId("ce-flow-thinking")).toBeInTheDocument();
|
||||
});
|
||||
|
||||
it("streams live working output (thinking + tools) while the agent works", () => {
|
||||
const session = makeSession({
|
||||
status: "active",
|
||||
currentQuestion: null,
|
||||
liveActivity: [
|
||||
{ kind: "thinking", text: "Considering options…", at: "t" },
|
||||
{ kind: "tool", text: "Grep", at: "t", done: false },
|
||||
],
|
||||
});
|
||||
render(<CeFlow session={session} onAnswer={vi.fn()} />);
|
||||
const pane = screen.getByTestId("ce-flow-live-activity");
|
||||
expect(pane).toHaveTextContent("Considering options…");
|
||||
expect(screen.getByTestId("ce-activity-tool")).toHaveTextContent("Grep");
|
||||
});
|
||||
|
||||
it("offers resume on an interrupted session", () => {
|
||||
const onResume = vi.fn();
|
||||
render(
|
||||
|
||||
@@ -70,14 +70,15 @@ export interface UseCeSessionResult {
|
||||
}
|
||||
|
||||
/**
|
||||
* Drive a single CE stage session through its lifecycle over the polling
|
||||
* routes: start → (poll while a turn runs) → render question → submit answer →
|
||||
* Drive a single CE stage session through its lifecycle: start → watch the
|
||||
* live working output while the turn runs → render question → submit answer →
|
||||
* continue → completed/error; resume an interrupted/error session.
|
||||
*
|
||||
* The session routes already run one turn synchronously per request and return
|
||||
* the post-turn state, so the common path settles immediately. Polling is the
|
||||
* fallback for a session left `active`/`launching` (e.g. recovered from another
|
||||
* process), honoring U5's client-polling transport.
|
||||
* Turn execution is DETACHED server-side: start/answer/resume return as soon
|
||||
* as the session row reflects the request (status `active`), and the client
|
||||
* converges via push (subscribe) with polling as the fallback. While a turn is
|
||||
* mid-flight, GET attaches `liveActivity` — the agent's streaming working
|
||||
* output — so each refetch updates the live pane.
|
||||
*/
|
||||
export function useCeSession(options: UseCeSessionOptions = {}): UseCeSessionResult {
|
||||
const transport = options.transport ?? defaultTransport;
|
||||
|
||||
@@ -7,16 +7,12 @@ import { asString } from "./route-helpers.js";
|
||||
/**
|
||||
* Session routes (U5): start / answer / resume / get-session-state.
|
||||
*
|
||||
* STREAMING TRANSPORT — HONEST STATEMENT.
|
||||
* Plugin routes return `{status, body}` with no native server-push; the loader
|
||||
* `emitEvent` is a logging stub, so there is no real plugin→client push path
|
||||
* today. v1 therefore uses POLLING: clients poll `GET /sessions/:id` for the
|
||||
* current persisted state (status, currentQuestion, conversationHistory). This
|
||||
* keeps U5 plugin-local and shippable and uses NO raw EventSource. The
|
||||
* orchestrator still emits observable events via `ctx.emitEvent` (a no-silent-
|
||||
* loss requirement); turning those into true client push needs a host
|
||||
* event-publish seam (publish-to-`/api/events`) — that is a carry-forward for
|
||||
* U6/follow-up, not faked here as push.
|
||||
* TRANSPORT. Turn execution is DETACHED: start/answer/resume return as soon
|
||||
* as the session row reflects the request, with the agent turn running in the
|
||||
* background. Clients converge via the `plugin:custom` SSE push (the
|
||||
* orchestrator emits throttled progress events) with `GET /sessions/:id`
|
||||
* polling as the fallback — that GET also attaches the in-flight working
|
||||
* output (`liveActivity`) so the user can watch the agent work mid-turn.
|
||||
*/
|
||||
|
||||
interface RouteRequest {
|
||||
@@ -62,8 +58,9 @@ export function createSessionRoutes(): PluginRouteDefinition[] {
|
||||
const result = await orch.start(stageId, {
|
||||
openingMessage,
|
||||
projectId: asString(body?.projectId) ?? null,
|
||||
detach: true,
|
||||
});
|
||||
return { status: 201, body: { session: result.session, event: result.event } };
|
||||
return { status: 201, body: { session: result.session } };
|
||||
} catch (err) {
|
||||
return { status: 400, body: { error: err instanceof Error ? err.message : String(err) } };
|
||||
}
|
||||
@@ -83,8 +80,10 @@ export function createSessionRoutes(): PluginRouteDefinition[] {
|
||||
|
||||
const orch = getOrchestrator(ctx);
|
||||
try {
|
||||
const result = await orch.answer(id, questionId, (body as Record<string, unknown>).response);
|
||||
return { status: 200, body: { session: result.session, event: result.event } };
|
||||
const result = await orch.answer(id, questionId, (body as Record<string, unknown>).response, {
|
||||
detach: true,
|
||||
});
|
||||
return { status: 200, body: { session: result.session } };
|
||||
} catch (err) {
|
||||
return { status: 409, body: { error: err instanceof Error ? err.message : String(err) } };
|
||||
}
|
||||
@@ -98,7 +97,7 @@ export function createSessionRoutes(): PluginRouteDefinition[] {
|
||||
const id = (req as RouteRequest).params.id;
|
||||
const orch = getOrchestrator(ctx);
|
||||
try {
|
||||
const result = await orch.resume(id);
|
||||
const result = await orch.resume(id, { detach: true });
|
||||
return { status: 200, body: { session: result.session } };
|
||||
} catch (err) {
|
||||
return { status: 404, body: { error: err instanceof Error ? err.message : String(err) } };
|
||||
@@ -108,12 +107,18 @@ export function createSessionRoutes(): PluginRouteDefinition[] {
|
||||
{
|
||||
method: "GET",
|
||||
path: "/sessions/:id",
|
||||
description: "Get current persisted session state (polling transport).",
|
||||
description: "Get current session state, including in-flight working output (liveActivity).",
|
||||
handler: async (req: unknown, ctx: PluginContext): Promise<PluginRouteResponse> => {
|
||||
const id = (req as RouteRequest).params.id;
|
||||
const session = getCeSessionStore(ctx).get(id);
|
||||
if (!session) return { status: 404, body: { error: `Session ${id} not found` } };
|
||||
return { status: 200, body: { session } };
|
||||
// Attach the orchestrator's transient mid-turn buffer so a polling
|
||||
// client can watch the agent work while the turn runs.
|
||||
const liveActivity = getOrchestrator(ctx).getLiveActivity(id);
|
||||
return {
|
||||
status: 200,
|
||||
body: { session: liveActivity.length > 0 ? { ...session, liveActivity } : session },
|
||||
};
|
||||
},
|
||||
},
|
||||
{
|
||||
|
||||
@@ -4,6 +4,7 @@ import type {
|
||||
CreateInteractiveAiSessionFactory,
|
||||
InteractiveAiSession,
|
||||
InteractiveAiSessionEvent,
|
||||
InteractiveAiSessionProgressEvent,
|
||||
PlanningQuestion,
|
||||
PluginContext,
|
||||
} from "@fusion/core";
|
||||
@@ -11,7 +12,7 @@ import { resolveDefaultInstallTargetRoot } from "../skill-installation.js";
|
||||
import { getCePipelineStore, type CePipelineStore } from "../sync/pipeline-store.js";
|
||||
import { createCeTaskWithLink } from "../sync/ce-task.js";
|
||||
import { getDefaultModelId, getDefaultProvider, getEnabledStages } from "../settings.js";
|
||||
import type { CeSession, CeSessionStore } from "./session-store.js";
|
||||
import type { CeActivityTurn, CeSession, CeSessionStore } from "./session-store.js";
|
||||
import { getCeSessionStore } from "./session-store.js";
|
||||
import { getStage, type CeStageDefinition } from "./stage-registry.js";
|
||||
|
||||
@@ -46,9 +47,24 @@ export interface CeDerivedTaskSpec {
|
||||
column?: string;
|
||||
}
|
||||
|
||||
/** Default per-turn timeout. A turn that exceeds this is treated as a stall. */
|
||||
/**
|
||||
* Default per-turn INACTIVITY timeout. A turn is treated as stalled only after
|
||||
* this long with NO live progress (thinking/text/tool activity) — a long but
|
||||
* actively-working turn is never killed. Without an onProgress-capable factory
|
||||
* (e.g. scripted test fakes), this degrades to a fixed per-turn timeout.
|
||||
*/
|
||||
const DEFAULT_TURN_TIMEOUT_MS = 120000;
|
||||
|
||||
/** Throttle for progress-driven SSE emits + lastActivityAt bumps. */
|
||||
const PROGRESS_EMIT_INTERVAL_MS = 500;
|
||||
|
||||
/** Caps so a runaway turn cannot grow the live buffer unbounded. */
|
||||
const MAX_ACTIVITY_TURNS = 200;
|
||||
const MAX_ACTIVITY_TURN_CHARS = 16000;
|
||||
/** Caps for the condensed activity trace persisted into history on settle. */
|
||||
const MAX_PERSISTED_ACTIVITY_TURNS = 50;
|
||||
const MAX_PERSISTED_ACTIVITY_TURN_CHARS = 4000;
|
||||
|
||||
/**
|
||||
* Observable event names emitted via `ctx.emitEvent`. The no-silent-loss
|
||||
* invariant requires that interrupt/error ALWAYS emit one of these AND persist
|
||||
@@ -64,17 +80,11 @@ export const CE_EVENTS = {
|
||||
|
||||
export class CeTurnTimeoutError extends Error {
|
||||
constructor(ms: number) {
|
||||
super(`CE session turn timed out after ${ms}ms`);
|
||||
super(`CE session turn stalled: no agent activity for ${ms}ms`);
|
||||
this.name = "CeTurnTimeoutError";
|
||||
}
|
||||
}
|
||||
|
||||
function timeoutAfter(ms: number): Promise<never> {
|
||||
return new Promise((_, reject) => {
|
||||
globalThis.setTimeout(() => reject(new CeTurnTimeoutError(ms)), ms).unref?.();
|
||||
});
|
||||
}
|
||||
|
||||
export interface OrchestratorDeps {
|
||||
ctx: PluginContext;
|
||||
/**
|
||||
@@ -122,6 +132,12 @@ export function buildStageSystemPrompt(stage: CeStageDefinition): string {
|
||||
' - To ask the user something: {"type":"question","data":{"id":"<unique>","type":"single_select|multi_select|text|confirm","question":"...","options":[{"id":"..","label":".."}]}}',
|
||||
' - When the stage is finished: {"type":"complete","data":{"artifact":"<full markdown document>", ...}}',
|
||||
"No markdown fences, no prose outside the JSON object.",
|
||||
"",
|
||||
"The user's reply arrives as {\"type\":\"answer\",\"questionId\":\"...\",\"response\":...}. The response takes one of three shapes:",
|
||||
" - a direct answer to your question (an option id, array of option ids, text, or boolean),",
|
||||
' - {"value": <direct answer>, "comment": "<guidance>"} — apply the answer AND incorporate the guidance into how you proceed,',
|
||||
' - {"feedback": "<guidance only>"} — the user is steering rather than answering. Incorporate the feedback, adjust course, and either re-ask the question (possibly revised) or continue if the feedback resolves it.',
|
||||
"Steering feedback is first-class input: never ignore it, and acknowledge course corrections in your next question or output.",
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
@@ -129,6 +145,12 @@ export interface StartStageOptions {
|
||||
/** Opening user message (the stage prompt / topic). */
|
||||
openingMessage: string;
|
||||
projectId?: string | null;
|
||||
/**
|
||||
* Return as soon as the session row exists, with the turn running in the
|
||||
* background (the route posture — lets clients watch live working output
|
||||
* via push/poll instead of blocking on the whole turn).
|
||||
*/
|
||||
detach?: boolean;
|
||||
}
|
||||
|
||||
/** Result of a single orchestrator step (start / answer / resume). */
|
||||
@@ -156,6 +178,14 @@ export class CeOrchestrator {
|
||||
private readonly turnTimeoutMs: number;
|
||||
/** Live in-memory session handles keyed by ce_session id. */
|
||||
private readonly live = new Map<string, InteractiveAiSession>();
|
||||
/** Mid-turn working output per session (transient; flushed to history on settle). */
|
||||
private readonly activity = new Map<string, CeActivityTurn[]>();
|
||||
/** Last progress timestamp per session (drives the inactivity watchdog). */
|
||||
private readonly lastProgressAt = new Map<string, number>();
|
||||
/** Last progress-driven emit per session (throttling). */
|
||||
private readonly lastProgressEmitAt = new Map<string, number>();
|
||||
/** Sessions currently REPLAYING history (rehydrate) — progress suppressed. */
|
||||
private readonly replaying = new Set<string>();
|
||||
|
||||
constructor(deps: OrchestratorDeps) {
|
||||
this.ctx = deps.ctx;
|
||||
@@ -174,7 +204,10 @@ export class CeOrchestrator {
|
||||
* bundled skill (closing the U2/U5 skill-discovery carry-forward). Model
|
||||
* provider/model are setting-gated (U9); omitted keys let the host pick defaults.
|
||||
*/
|
||||
private buildSessionOptions(stage: CeStageDefinition): Parameters<CreateInteractiveAiSessionFactory>[0] {
|
||||
private buildSessionOptions(
|
||||
stage: CeStageDefinition,
|
||||
sessionId: string,
|
||||
): Parameters<CreateInteractiveAiSessionFactory>[0] {
|
||||
const defaultProvider = getDefaultProvider(this.ctx.settings);
|
||||
const defaultModelId = getDefaultModelId(this.ctx.settings);
|
||||
return {
|
||||
@@ -183,12 +216,128 @@ export class CeOrchestrator {
|
||||
tools: "coding",
|
||||
requestedSkillNames: [stage.skillId],
|
||||
additionalSkillPaths: resolveStageSkillPaths(),
|
||||
onProgress: (event) => this.handleProgress(sessionId, event),
|
||||
...(defaultProvider ? { defaultProvider } : {}),
|
||||
...(defaultModelId ? { defaultModelId } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
/** Start a fresh session for a registered stage and run the opening turn. */
|
||||
/**
|
||||
* Live mid-turn visibility. Accumulates streamed deltas into the session's
|
||||
* activity buffer (consecutive deltas of one kind merge into one turn; tool
|
||||
* markers are discrete), pokes the inactivity watchdog, and — throttled —
|
||||
* bumps the persisted liveness anchor and emits an observable turn event so
|
||||
* push clients refetch. Replay (rehydrate) progress is fully suppressed:
|
||||
* it reconstructs context, it is not new work.
|
||||
*/
|
||||
private handleProgress(sessionId: string, event: InteractiveAiSessionProgressEvent): void {
|
||||
if (this.replaying.has(sessionId)) return;
|
||||
this.lastProgressAt.set(sessionId, Date.now());
|
||||
|
||||
const turns = this.activity.get(sessionId) ?? [];
|
||||
if (!this.activity.has(sessionId)) this.activity.set(sessionId, turns);
|
||||
const now = new Date().toISOString();
|
||||
if (event.type === "tool") {
|
||||
if (event.phase === "start") {
|
||||
turns.push({ kind: "tool", text: event.name, at: now, done: false });
|
||||
} else {
|
||||
// Mark the most recent still-open tool turn with this name as done.
|
||||
for (let i = turns.length - 1; i >= 0; i--) {
|
||||
const t = turns[i];
|
||||
if (t.kind === "tool" && t.text === event.name && !t.done) {
|
||||
t.done = true;
|
||||
if (event.isError) t.isError = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
const last = turns[turns.length - 1];
|
||||
if (last && last.kind === event.type) {
|
||||
if (last.text.length < MAX_ACTIVITY_TURN_CHARS) {
|
||||
last.text = (last.text + event.delta).slice(0, MAX_ACTIVITY_TURN_CHARS);
|
||||
}
|
||||
} else {
|
||||
turns.push({ kind: event.type, text: event.delta.slice(0, MAX_ACTIVITY_TURN_CHARS), at: now });
|
||||
}
|
||||
}
|
||||
// Cap the buffer; drop oldest (the tail is what the user is watching).
|
||||
if (turns.length > MAX_ACTIVITY_TURNS) turns.splice(0, turns.length - MAX_ACTIVITY_TURNS);
|
||||
|
||||
const nowMs = Date.now();
|
||||
if (nowMs - (this.lastProgressEmitAt.get(sessionId) ?? 0) >= PROGRESS_EMIT_INTERVAL_MS) {
|
||||
this.lastProgressEmitAt.set(sessionId, nowMs);
|
||||
// Bump lastActivityAt so the staleness rubric sees an actively-working
|
||||
// turn as alive; emit so push clients refetch (GET attaches the buffer).
|
||||
this.store.update(sessionId, {});
|
||||
this.ctx.emitEvent(CE_EVENTS.turn, { sessionId, kind: "progress" });
|
||||
}
|
||||
}
|
||||
|
||||
/** Read the in-flight working output for a session (route accessor). */
|
||||
getLiveActivity(sessionId: string): CeActivityTurn[] {
|
||||
return this.activity.get(sessionId) ?? [];
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist a condensed copy of the live activity buffer into history (so the
|
||||
* transcript keeps the working trace after the turn settles), then clear it.
|
||||
*/
|
||||
private flushActivity(sessionId: string): void {
|
||||
const turns = this.activity.get(sessionId);
|
||||
this.activity.delete(sessionId);
|
||||
if (!turns || turns.length === 0) return;
|
||||
const condensed = turns.slice(-MAX_PERSISTED_ACTIVITY_TURNS).map((t) => ({
|
||||
...t,
|
||||
text: t.text.slice(0, MAX_PERSISTED_ACTIVITY_TURN_CHARS),
|
||||
}));
|
||||
this.store.appendHistory(sessionId, {
|
||||
role: "agent",
|
||||
text: JSON.stringify({ activity: { turns: condensed } }),
|
||||
at: new Date().toISOString(),
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Inactivity watchdog: rejects only after `turnTimeoutMs` with NO progress.
|
||||
* Every progress event re-arms it, so a long actively-working turn survives;
|
||||
* with a non-streaming factory it degrades to a fixed per-turn timeout.
|
||||
*/
|
||||
private createWatchdog(sessionId: string): { promise: Promise<never>; cancel(): void } {
|
||||
let cancelled = false;
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
this.lastProgressAt.set(sessionId, Date.now());
|
||||
const promise = new Promise<never>((_, reject) => {
|
||||
const check = () => {
|
||||
if (cancelled) return;
|
||||
const elapsed = Date.now() - (this.lastProgressAt.get(sessionId) ?? 0);
|
||||
if (elapsed >= this.turnTimeoutMs) {
|
||||
reject(new CeTurnTimeoutError(this.turnTimeoutMs));
|
||||
return;
|
||||
}
|
||||
timer = globalThis.setTimeout(check, this.turnTimeoutMs - elapsed);
|
||||
timer.unref?.();
|
||||
};
|
||||
check();
|
||||
});
|
||||
return {
|
||||
promise,
|
||||
cancel: () => {
|
||||
cancelled = true;
|
||||
if (timer) clearTimeout(timer);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Start a fresh session for a registered stage and run the opening turn.
|
||||
*
|
||||
* `detach: true` (the route posture) returns as soon as the session row
|
||||
* exists with the turn running in the background — the client converges via
|
||||
* push/poll and can watch the live working output. Errors during a detached
|
||||
* turn surface through session state (failSession/interruptSession), never
|
||||
* as an unhandled rejection. Validation errors still throw synchronously.
|
||||
*/
|
||||
async start(stageId: string, opts: StartStageOptions): Promise<CeStepResult> {
|
||||
const stage = getStage(stageId);
|
||||
if (!stage) throw new Error(`Unknown CE stage: ${stageId}`);
|
||||
@@ -202,27 +351,46 @@ export class CeOrchestrator {
|
||||
);
|
||||
}
|
||||
|
||||
let session = this.store.create({
|
||||
const session = this.store.create({
|
||||
stage: stageId,
|
||||
projectId: opts.projectId ?? null,
|
||||
turnIntervalMs: this.turnTimeoutMs,
|
||||
});
|
||||
this.store.appendHistory(session.id, { role: "user", text: opts.openingMessage, at: new Date().toISOString() });
|
||||
|
||||
let interactive;
|
||||
try {
|
||||
interactive = await this.factory(this.buildSessionOptions(stage));
|
||||
} catch (err) {
|
||||
return { session: this.failSession(session.id, err), event: undefined };
|
||||
const turn = this.runOpeningTurn(session.id, stage, opts.openingMessage);
|
||||
if (opts.detach) {
|
||||
// runOpeningTurn never rejects (all failures persist into session state).
|
||||
void turn;
|
||||
return { session: this.requireSession(session.id) };
|
||||
}
|
||||
this.live.set(session.id, interactive.session);
|
||||
session = this.store.update(session.id, { status: "active" }) ?? session;
|
||||
|
||||
return this.runTurn(session.id, () => interactive.session.prompt(opts.openingMessage), interactive.session);
|
||||
return turn;
|
||||
}
|
||||
|
||||
/** Answer the awaiting question and continue the loop. */
|
||||
async answer(sessionId: string, questionId: string, response: unknown): Promise<CeStepResult> {
|
||||
/** Create the live handle and run the opening turn. Never rejects. */
|
||||
private async runOpeningTurn(
|
||||
sessionId: string,
|
||||
stage: CeStageDefinition,
|
||||
openingMessage: string,
|
||||
): Promise<CeStepResult> {
|
||||
let interactive;
|
||||
try {
|
||||
interactive = await this.factory!(this.buildSessionOptions(stage, sessionId));
|
||||
} catch (err) {
|
||||
return { session: this.failSession(sessionId, err), event: undefined };
|
||||
}
|
||||
this.live.set(sessionId, interactive.session);
|
||||
this.store.update(sessionId, { status: "active" });
|
||||
return this.runTurn(sessionId, () => interactive.session.prompt(openingMessage), interactive.session);
|
||||
}
|
||||
|
||||
/** Answer the awaiting question and continue the loop (detachable like start). */
|
||||
async answer(
|
||||
sessionId: string,
|
||||
questionId: string,
|
||||
response: unknown,
|
||||
opts: { detach?: boolean } = {},
|
||||
): Promise<CeStepResult> {
|
||||
const session = this.requireSession(sessionId);
|
||||
if (session.status !== "awaiting_input") {
|
||||
throw new Error(`Session ${sessionId} is not awaiting input (status=${session.status}).`);
|
||||
@@ -247,7 +415,13 @@ export class CeOrchestrator {
|
||||
at: new Date().toISOString(),
|
||||
});
|
||||
this.store.update(sessionId, { status: "active", currentQuestion: null });
|
||||
return this.runTurn(sessionId, () => live.answer(questionId, response), live);
|
||||
const turn = this.runTurn(sessionId, () => live.answer(questionId, response), live);
|
||||
if (opts.detach) {
|
||||
// runTurn never rejects (all failures persist into session state).
|
||||
void turn;
|
||||
return { session: this.requireSession(sessionId) };
|
||||
}
|
||||
return turn;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -268,7 +442,7 @@ export class CeOrchestrator {
|
||||
* process), we DO NOT advertise a misleading answerable status: the session is
|
||||
* left `interrupted` with a clear error explaining it can't be continued here.
|
||||
*/
|
||||
async resume(sessionId: string): Promise<CeStepResult> {
|
||||
async resume(sessionId: string, opts: { detach?: boolean } = {}): Promise<CeStepResult> {
|
||||
const session = this.requireSession(sessionId);
|
||||
|
||||
// Terminal / already-answerable-with-a-live-handle cases need no rehydration.
|
||||
@@ -305,17 +479,27 @@ export class CeOrchestrator {
|
||||
return { session: next };
|
||||
}
|
||||
|
||||
try {
|
||||
await this.rehydrate(session);
|
||||
} catch (err) {
|
||||
// Rehydration failed — keep progress, surface the failure, do not advertise
|
||||
// an answerable status we can't back.
|
||||
const next = this.interruptSession(sessionId, err);
|
||||
const rehydration = (async (): Promise<CeStepResult> => {
|
||||
try {
|
||||
await this.rehydrate(session);
|
||||
} catch (err) {
|
||||
// Rehydration failed — keep progress, surface the failure, do not
|
||||
// advertise an answerable status we can't back.
|
||||
return { session: this.interruptSession(sessionId, err) };
|
||||
}
|
||||
const next = this.store.update(sessionId, { status: "awaiting_input", error: null }) ?? session;
|
||||
return { session: next };
|
||||
})();
|
||||
|
||||
if (opts.detach) {
|
||||
// Rehydration replays the conversation against the live model and can be
|
||||
// slow; the route posture marks the session active and converges via
|
||||
// push/poll. The IIFE never rejects (failures persist into state).
|
||||
const next = this.store.update(sessionId, { status: "active", error: null }) ?? session;
|
||||
void rehydration;
|
||||
return { session: next };
|
||||
}
|
||||
|
||||
const next = this.store.update(sessionId, { status: "awaiting_input", error: null }) ?? session;
|
||||
return { session: next };
|
||||
return rehydration;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -329,7 +513,18 @@ export class CeOrchestrator {
|
||||
const stage = getStage(session.stage);
|
||||
if (!stage) throw new Error(`Unknown CE stage: ${session.stage}`);
|
||||
|
||||
const interactive = await this.factory!(this.buildSessionOptions(stage));
|
||||
// Replay is side-effect-suppressed — including live progress, which would
|
||||
// otherwise re-stream the old turns' output as if it were new work.
|
||||
this.replaying.add(session.id);
|
||||
try {
|
||||
await this.rehydrateReplay(session, stage);
|
||||
} finally {
|
||||
this.replaying.delete(session.id);
|
||||
}
|
||||
}
|
||||
|
||||
private async rehydrateReplay(session: CeSession, stage: CeStageDefinition): Promise<void> {
|
||||
const interactive = await this.factory!(this.buildSessionOptions(stage, session.id));
|
||||
const live = interactive.session;
|
||||
|
||||
// Walk the recorded user turns in order. The FIRST user turn is the opening
|
||||
@@ -407,21 +602,24 @@ export class CeOrchestrator {
|
||||
live: InteractiveAiSession,
|
||||
): Promise<CeStepResult> {
|
||||
let event: InteractiveAiSessionEvent;
|
||||
const watchdog = this.createWatchdog(sessionId);
|
||||
try {
|
||||
event = await Promise.race([
|
||||
(async () => {
|
||||
await driver();
|
||||
return live.nextEvent();
|
||||
})(),
|
||||
timeoutAfter(this.turnTimeoutMs),
|
||||
watchdog.promise,
|
||||
]);
|
||||
} catch (err) {
|
||||
// Timeout or driver throw → auto-save as interrupted (progress preserved)
|
||||
// and emit an observable event. Never silent loss.
|
||||
watchdog.cancel();
|
||||
const session = this.interruptSession(sessionId, err);
|
||||
this.disposeLive(sessionId);
|
||||
return { session, event: { type: "error", data: { message: session.error ?? "interrupted", cause: err } } };
|
||||
}
|
||||
watchdog.cancel();
|
||||
|
||||
const session = this.applyEvent(sessionId, event);
|
||||
if (event.type === "complete" || event.type === "error") {
|
||||
@@ -438,6 +636,11 @@ export class CeOrchestrator {
|
||||
|
||||
/** Persist a seam event onto the session row + emit the matching observable event. */
|
||||
private applyEvent(sessionId: string, event: InteractiveAiSessionEvent): CeSession {
|
||||
// The turn settled — persist its working trace into history (so the
|
||||
// transcript keeps it) BEFORE the settling record, then clear the buffer.
|
||||
if (event.type === "question" || event.type === "complete" || event.type === "error") {
|
||||
this.flushActivity(sessionId);
|
||||
}
|
||||
switch (event.type) {
|
||||
case "thinking":
|
||||
case "text": {
|
||||
@@ -547,6 +750,9 @@ export class CeOrchestrator {
|
||||
|
||||
/** Persist `interrupted` with progress preserved and emit. */
|
||||
private interruptSession(sessionId: string, cause: unknown): CeSession {
|
||||
// Keep the working trace: an interrupted turn's output is exactly what the
|
||||
// user needs to see to understand where it stopped.
|
||||
this.flushActivity(sessionId);
|
||||
const message = cause instanceof Error ? cause.message : String(cause);
|
||||
const s =
|
||||
this.store.update(sessionId, { status: "interrupted", error: message }) ?? this.requireSession(sessionId);
|
||||
@@ -606,5 +812,8 @@ export class CeOrchestrator {
|
||||
}
|
||||
this.live.delete(sessionId);
|
||||
}
|
||||
this.activity.delete(sessionId);
|
||||
this.lastProgressAt.delete(sessionId);
|
||||
this.lastProgressEmitAt.delete(sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -33,12 +33,32 @@ export interface CeConversationTurn {
|
||||
at: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* One line of live agent activity (mid-turn working output): an accumulated
|
||||
* thinking/text block or a discrete tool execution marker.
|
||||
*/
|
||||
export interface CeActivityTurn {
|
||||
kind: "thinking" | "text" | "tool";
|
||||
text: string;
|
||||
at: string;
|
||||
/** Tool turns: execution finished. */
|
||||
done?: boolean;
|
||||
/** Tool turns: execution finished with an error. */
|
||||
isError?: boolean;
|
||||
}
|
||||
|
||||
export interface CeSession {
|
||||
id: string;
|
||||
stage: string;
|
||||
status: CeSessionStatus;
|
||||
currentQuestion: PlanningQuestion | null;
|
||||
conversationHistory: CeConversationTurn[];
|
||||
/**
|
||||
* TRANSIENT: in-flight working output for the current turn, attached by the
|
||||
* GET-session route from the orchestrator's in-memory buffer. Never persisted
|
||||
* to the row; absent when no turn is running (or in another process).
|
||||
*/
|
||||
liveActivity?: CeActivityTurn[];
|
||||
projectId: string | null;
|
||||
artifactPath: string | null;
|
||||
error: string | null;
|
||||
|
||||
Reference in New Issue
Block a user