Skip to content
15 changes: 15 additions & 0 deletions changelog.d/fixes/orchestration-followups.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
- **fix(dashboard):** five follow-ups from the Phase 2 Orchestration Canvas review. The History
drawer now shows the memory section for runs that already left the live TTL window: the
persisted `memory_hits` event is parsed into `metadata.memoryHits` with the same defensive
validation the drawer applies, and — because that event is observability rather than a state
transition — it no longer leaks into the timeline, where it had been inheriting the task's
state and rendering as a duplicate transition. The A2A memory recall runs against its own 1.5s
deadline instead of inheriting the memory backend's 30s one; overshooting degrades exactly like
any other recall failure (no hits, task proceeds), and the timer is cleared on both paths.
Repeating a Conductor run carries its `requirements.cli`/`requirements.model` forward, so the
new run is pinned to the same runner profile and model rather than drifting to whatever the
fleet picks. A successful repeat from the Agents tab now focuses the run it created, instead of
leaving the operator on the finished one. And the auth test for `POST /api/conductor/tasks`
moved into the shared `ROUTES` array rather than restating the pattern.

Closes #12639
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,13 @@ export default function OrchestrationPageClient() {
[collapsed, setParams]
);
const closeDrawer = useCallback(() => setParams({ node: null }), [setParams]);
const onActionDone = useCallback(
(newNodeId?: string) => {
refetch();
if (newNodeId) setParams({ node: newNodeId });
},
[refetch, setParams]
);
const clearFilters = useCallback(
() => setParams({ q: null, state: null, source: null, provider: null }),
[setParams]
Expand Down Expand Up @@ -176,8 +183,17 @@ export default function OrchestrationPageClient() {
{tab === "history" && <HistoryTab />}
</div>
</div>
{/* A successful repeat hands back the CANVAS id of the task it created: refetch, then
focus it (`?node=`) so the operator lands on the new run instead of staring at the
finished one. No id (approve/cancel, or a creation response without one) keeps the
current selection. The History tab renders its own drawer and deliberately does NOT
navigate (HistoryTab.tsx) — its runs are not addressable in the live snapshot. */}
{tab !== "history" && (
<OrchestrationDrawer node={selectedNode} onClose={closeDrawer} onActionDone={refetch} />
<OrchestrationDrawer
node={selectedNode}
onClose={closeDrawer}
onActionDone={onActionDone}
/>
)}
</div>
);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ import { useTranslations } from "next-intl";
import { StatusDot } from "@/shared/components/flow/StatusDot";
import { orchStateColor, type OrchNode, type OrchState } from "../model/orchestrationTypes";
import { useDrawerDetail } from "./useDrawerDetail";
import type { DrawerError } from "./useDrawerDetail";
import type { DrawerError, RepeatOutcome } from "./useDrawerDetail";
import type { CloudAgentTask } from "@/lib/cloudAgent/types";
import type { A2ATask } from "@/lib/a2a/taskManager";

Expand Down Expand Up @@ -349,15 +349,18 @@ function RepeatButton({
}: {
canRepeat: boolean;
busy: boolean;
repeat: () => Promise<boolean>;
onActionDone: () => void;
repeat: () => Promise<RepeatOutcome>;
onActionDone: (newNodeId?: string) => void;
onToast: (text: string) => void;
t: Translate;
}) {
const { confirming, onClick } = useTwoClickConfirm(() => {
void (async () => {
if (await repeat()) {
onActionDone();
// The created task's CANVAS id is handed to `onActionDone` so the page can focus it —
// `undefined` when the creation response carried no usable id (plain refetch, no jump).
const { ok, newNodeId } = await repeat();
if (ok) {
onActionDone(newNodeId ?? undefined);
onToast(t("repeatDone"));
}
})();
Expand Down Expand Up @@ -395,8 +398,8 @@ function DrawerActions({
busy: boolean;
approve: () => Promise<boolean>;
cancel: () => Promise<boolean>;
repeat: () => Promise<boolean>;
onActionDone: () => void;
repeat: () => Promise<RepeatOutcome>;
onActionDone: (newNodeId?: string) => void;
onToast: (text: string) => void;
t: Translate;
}) {
Expand Down Expand Up @@ -488,7 +491,7 @@ export function OrchestrationDrawer({
}: {
node: OrchNode | null;
onClose: () => void;
onActionDone: () => void;
onActionDone: (newNodeId?: string) => void;
}) {
const t = useTranslations("orchestration");
const {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,14 @@ function repeatReqForA2a(
};
}

/** conductor repeat builder — `POST /api/conductor/tasks` (D1 task-creation route). */
/**
* conductor repeat builder — `POST /api/conductor/tasks` (D1 task-creation route).
* `cli`/`model` come from the hub's `requirements` (`ConductorTaskDetail`, hubProxy.ts) and are
* carried over so the repeat lands on the SAME runner profile/model the original task was
* pinned to. Both are `z.string().optional()` in the route's Zod: a `null` would 400, so a
* missing requirement must OMIT the field (`undefined`) rather than send `null` — and the two
* are independent (one may be set while the other is not).
*/
function repeatReqForConductor(detail: unknown): { url: string; init: RequestInit } | null {
const d = detail as ConductorTaskDetail | null;
if (!d?.repo || !d?.prompt) return null;
Expand All @@ -154,6 +161,8 @@ function repeatReqForConductor(detail: unknown): { url: string; init: RequestIni
prompt: d.prompt,
baseRef: d.base_ref ?? undefined,
mode: d.mode,
cli: d.cli ?? undefined,
model: d.model ?? undefined,
}),
};
}
Expand All @@ -168,6 +177,36 @@ export function repeatReqFor(
return null;
}

/** `<prefix><id>` when `id` is a non-empty string, `null` otherwise (never a bare prefix). */
function prefixedNodeId(prefix: string, id: unknown): string | null {
return typeof id === "string" && id.length > 0 ? `${prefix}${id}` : null;
}

/**
* CANVAS node id of the task a successful repeat just created, from the creation response
* body — `null` whenever the body does not carry a usable id (the caller then simply refetches
* without focusing anything). The canvas addresses nodes by PREFIXED id
* (`mergeSnapshot.ts`), so the raw upstream id is never returned on its own. Response
* envelopes, verified against the live routes:
* - conductor (`POST /api/conductor/tasks`): `{ task_id }`.
* - cloud-agent (`POST /api/v1/agents/tasks`): `{ data: { id } }`.
* - a2a (`POST /a2a`, JSON-RPC `message/send`): `{ result: { task: { id } } }`.
*/
export function newNodeIdFrom(node: OrchNode, body: unknown): string | null {
if (!body || typeof body !== "object") return null;
const b = body as Record<string, unknown>;
if (node.id.startsWith("conductor:task:")) return prefixedNodeId("conductor:task:", b.task_id);
if (node.id.startsWith("cloud-agent:")) {
const data = b.data as { id?: unknown } | undefined;
return prefixedNodeId("cloud-agent:", data?.id);
}
if (node.id.startsWith("a2a:")) {
const result = b.result as { task?: { id?: unknown } } | undefined;
return prefixedNodeId("a2a:", result?.task?.id);
}
return null;
}

/**
* Unwraps a task-detail GET response to the actual task payload. Each source's
* route has its own envelope — verified against the live handlers, not assumed:
Expand Down Expand Up @@ -258,43 +297,62 @@ function useFetchDetail(
* that never happened, so the `/a2a` action also inspects the envelope. Only the numeric
* `error.code` is surfaced (`RPC <code>`) — never the upstream `error.message`.
*/
async function jsonRpcErrorCode(res: {
json?: () => Promise<unknown>;
}): Promise<number | undefined> {
function jsonRpcErrorCode(body: unknown): number | undefined {
const b = body as { error?: { code?: unknown } } | undefined | null;
const code = b?.error?.code;
return typeof code === "number" ? code : b?.error ? -32603 : undefined;
}

/**
* Reads an action response body ONCE, tolerating a non-JSON/empty body. A body that cannot be
* parsed is not evidence of failure — the status already stood — so it yields `null` and the
* action stays successful (it just has no new-task id to focus).
*/
async function readJsonBody(res: { json?: () => Promise<unknown> }): Promise<unknown> {
try {
const body = (await res.json?.()) as { error?: { code?: unknown } } | undefined;
const code = body?.error?.code;
return typeof code === "number" ? code : body?.error ? -32603 : undefined;
return (await res.json?.()) ?? null;
} catch {
// A non-JSON / already-consumed body is not evidence of failure — the status stands.
return undefined;
return null;
}
}

/** Outcome of an action POST: whether it succeeded, plus the parsed body on success. */
interface ActionOutcome {
ok: boolean;
body: unknown;
}

async function performAction(
req: { url: string; init: RequestInit } | null,
setActionError: (text: string) => void,
clearError: () => void
): Promise<boolean> {
if (!req) return false;
): Promise<ActionOutcome> {
if (!req) return { ok: false, body: null };
try {
const res = await fetch(req.url, req.init);
if (!res.ok) throw new Error(`HTTP ${res.status}`);
const body = await readJsonBody(res);
if (req.url === "/a2a") {
const code = await jsonRpcErrorCode(res);
const code = jsonRpcErrorCode(body);
if (code !== undefined) throw new Error(`RPC ${code}`);
}
// The banner is not sticky: a retry (or any later action) that works clears whatever
// detail/action error was on screen, so the drawer never shows a failure the operator
// already recovered from.
clearError();
return true;
return { ok: true, body };
} catch (err) {
setActionError(toSafeErrorText(err));
return false;
return { ok: false, body: null };
}
}

/** Result of the drawer's repeat action: success plus the canvas id of the created task. */
export interface RepeatOutcome {
ok: boolean;
newNodeId: string | null;
}

export function useDrawerDetail(node: OrchNode | null) {
const [detail, setDetail] = useState<unknown | null>(null);
const [isLoading, setIsLoading] = useState(false);
Expand All @@ -311,15 +369,19 @@ export function useDrawerDetail(node: OrchNode | null) {
const { canApprove, canCancel } = deriveActionAvailability(route, node);
const repeatReq = node ? repeatReqFor(node, detail) : null;

const runAction = async (req: { url: string; init: RequestInit } | null): Promise<boolean> => {
if (busy) return false;
const runAction = async (
req: { url: string; init: RequestInit } | null
): Promise<ActionOutcome> => {
if (busy) return { ok: false, body: null };
setBusy(true);
try {
return await performAction(req, setActionError, () => setErrorState(null));
} finally {
setBusy(false);
}
};
const runBooleanAction = async (req: { url: string; init: RequestInit } | null) =>
(await runAction(req)).ok;

return {
detail,
Expand All @@ -330,8 +392,13 @@ export function useDrawerDetail(node: OrchNode | null) {
canApprove,
canCancel,
canRepeat: !!repeatReq && !busy,
approve: () => runAction(route?.approveReq ?? null),
cancel: () => runAction(route?.cancelReq ?? null),
repeat: () => runAction(repeatReq),
approve: () => runBooleanAction(route?.approveReq ?? null),
cancel: () => runBooleanAction(route?.cancelReq ?? null),
// Only the repeat reports a new node id: approve/cancel act on the task already open, so
// there is nothing new to focus (and their responses can echo the SAME task's id back).
repeat: async (): Promise<RepeatOutcome> => {
const { ok, body } = await runAction(repeatReq);
return { ok, newNodeId: ok && node ? newNodeIdFrom(node, body) : null };
},
};
}
76 changes: 60 additions & 16 deletions src/app/api/a2a/tasks/[id]/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,40 @@ function safeJsonParse<T>(json: string | null | undefined, fallback: T): T {
}

const STATE_EVENT_PREFIX = "state:";
/** Event type written by `collectMemoryHits` (src/lib/a2a/taskExecution.ts). */
const MEMORY_HITS_EVENT_TYPE = "memory_hits";
const MEMORY_HIT_FIELDS = ["id", "key", "type", "snippet"] as const;

interface MemoryHit {
id: string;
key: string;
type: string;
snippet: string;
}

/**
* Parse a persisted `memory_hits` event's `data_json` into the hits the drawer renders.
* Mirrors `DrawerMemory`'s validation in
* `src/app/(dashboard)/dashboard/orchestration/drawer/OrchestrationDrawer.tsx`: `metadata` is
* caller-supplied and unvalidated end to end, so what got persisted can be anything —
* a bare string (`"boom"`, whose `.length` is truthy), an object, or an array carrying entries
* with a non-string `key`/`type`/`snippet` (rendered as React children, so an object there
* would throw "Objects are not valid as a React child" and take the whole drawer down).
* Never throws: a malformed payload degrades to an empty list, so the route answers 200
* with `metadata: {}` instead of 500.
*/
function parseMemoryHits(dataJson: string | null | undefined): MemoryHit[] {
const raw = safeJsonParse<unknown>(dataJson, null);
if (!Array.isArray(raw)) return [];
return raw.filter(
(hit): hit is MemoryHit =>
!!hit &&
typeof hit === "object" &&
MEMORY_HIT_FIELDS.every(
(field) => typeof (hit as Record<string, unknown>)[field] === "string"
)
);
}

/**
* Reconstitute the in-memory `A2ATask` shape (src/lib/a2a/taskManager.ts) from a persisted
Expand All @@ -29,22 +63,32 @@ const STATE_EVENT_PREFIX = "state:";
* state each event represents is recovered by stripping that prefix.
*/
function reconstituteHistoricalTask(row: A2ATaskHistoryRow) {
const input = safeJsonParse<{ skill: string; messages: Array<{ role: string; content: string }> }>(
row.input_json,
{ skill: row.skill_id ?? "", messages: [] }
);
const input = safeJsonParse<{
skill: string;
messages: Array<{ role: string; content: string }>;
}>(row.input_json, { skill: row.skill_id ?? "", messages: [] });
const artifacts = safeJsonParse<unknown[]>(row.output_json, []);
const events = listA2ATaskEvents(row.id).map((event) => {
const data = safeJsonParse<{ message?: string } | null>(event.data_json, null);
const state = event.event_type.startsWith(STATE_EVENT_PREFIX)
? event.event_type.slice(STATE_EVENT_PREFIX.length)
: row.state;
return {
timestamp: event.created_at,
state,
...(data?.message !== undefined ? { message: data.message } : {}),
};
});
const eventRows = listA2ATaskEvents(row.id);

// `memory_hits` is observability, not a state transition — it is hydrated into `metadata`
// (Fase 3, Task C1) and kept out of the timeline the drawer renders.
const memoryHits = eventRows
.filter((event) => event.event_type === MEMORY_HITS_EVENT_TYPE)
.flatMap((event) => parseMemoryHits(event.data_json));

const events = eventRows
.filter((event) => event.event_type !== MEMORY_HITS_EVENT_TYPE)
.map((event) => {
const data = safeJsonParse<{ message?: string } | null>(event.data_json, null);
const state = event.event_type.startsWith(STATE_EVENT_PREFIX)
? event.event_type.slice(STATE_EVENT_PREFIX.length)
: row.state;
return {
timestamp: event.created_at,
state,
...(data?.message !== undefined ? { message: data.message } : {}),
};
});

return {
id: row.id,
Expand All @@ -53,7 +97,7 @@ function reconstituteHistoricalTask(row: A2ATaskHistoryRow) {
input,
artifacts,
events,
metadata: {},
metadata: memoryHits.length > 0 ? { memoryHits } : {},
createdAt: row.created_at,
updatedAt: row.updated_at,
expiresAt: row.updated_at,
Expand Down
Loading
Loading