FN-7402: publish workflow step activity runs
Publish workflow step lifecycle activity so dashboard analytics can count graph-owned execution. - Emit best-effort agentRuns rows when workflow step sessions start and finish. - Include task lineage, step metadata, terminal status, and token usage in workflow step activity records. - Cover workflow step activity analytics and document dashboard Activity/Overview counting behavior. Files changed: .changeset/fn-7402-workflow-step-activity.md | 7 + docs/dashboard-guide.md | 5 +- .../core/src/__tests__/activity-analytics.test.ts | 44 +++++++ .../src/__tests__/step-session-executor.test.ts | 145 +++++++++++++++++++++ packages/engine/src/step-session-executor.ts | 104 ++++++++++++++- 5 files changed, 300 insertions(+), 5 deletions(-) Fusion-Task-Id: FN-7402 Fusion-Task-Lineage: ecb2810d-da05-4fd0-a3bd-da72cae99a31 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-7402-workflow-step-activity.md
Normal file
7
.changeset/fn-7402-workflow-step-activity.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Include graph-owned workflow step execution in Command Center activity analytics.
|
||||
category: fix
|
||||
dev: StepSessionExecutor now publishes best-effort agentRuns lifecycle rows for workflow step sessions.
|
||||
@@ -928,13 +928,14 @@ Features:
|
||||
<!-- FNXC:CommandCenter 2026-06-26-00:00: The four Overview Concurrency sliders change live scheduler capacity, so each settled edit opens a confirmation popup before persisting; cancel, backdrop, or Escape leaves the previous persisted value in place. -->
|
||||
<!-- FNXC:CommandCenter 2026-06-27-10:03: Tokens detail charts must show every model bucket returned by analytics for accurate spend attribution; Overview remains a compact top-model summary because its copy explicitly frames those cards as top consumers/share. -->
|
||||
<!-- FNXC:CommandCenterActivity 2026-06-30-00:00: Activity active-agent counts include both durable-agent usage events and ephemeral task-worker execution runs from agentRuns, because task execution can be visible without a matching usage_events row. -->
|
||||
<!-- FNXC:CommandCenterActivity 2026-07-01-00:00: Graph-owned workflow step sessions publish active-to-terminal agentRuns lifecycle rows with task lineage and step metadata, so daily activity and Activity throughput charts include new workflow execution without dashboard-side recounting. -->
|
||||
- **Overview controls dashboard** sits at the top of the Overview landing surface on desktop and mobile. It includes AI engine stop/start backed by `globalPause`, live scheduler status from executor stats, the shared Global Max Concurrent slider backed by `/api/global-concurrency`, range sliders for `maxConcurrent`, `maxTriageConcurrent`, and `maxWorktrees` that persist through `/api/settings`, and a compact theme dropdown with the same color-chip swatches and Shadcn variant list as Settings → Appearance. The four concurrency sliders ask for confirmation after a changed value settles; confirming persists the new cap, while cancel, backdrop, or Escape dismissal reverts to the last persisted value without saving. The global and current-project max-concurrent sliders show running-agent counts plus a current-use dot on the track once utilization data loads; triage and worktree sliders remain cap-only. These controls reuse existing APIs and App-level theme setters; they do not add a new backend route or second theme owner.
|
||||
- **Overview** summarizes token usage/cost, autonomy, active nodes, sessions, agent runs, tasks done, model breadth, and real open signals, and includes the SDLC throughput funnel for the selected range at the bottom of the Overview content in loading, error, empty, and populated states. Its token total and Live activity snapshot token metric refresh on a bounded live cadence and animate number changes while preserving reduced-motion preferences. The sessions card uses the selected-range `ActivityAnalytics.sessions` value already loaded for the overview. The Live activity snapshot also shows the current board-state count for tasks in progress, independent of the selected analytics date range. Overview's active-agent and daily activity values count durable-agent usage events plus ephemeral task-worker execution runs, with the same agent counted once per day/range if both sources record activity. Overview includes a graph-rich software-factory snapshot with the existing top-model-consumers bar, tool-category bar, top-model token-share pie, and the daily activity multi-series line chart placed before the daily activity sparkline/trend so the richer line graph sits higher in the chart grid. These reuse the already-loaded tokens, tools, activity, and signals analytics; the signals count comes from `/api/command-center/signals` and renders unavailable (`—`) while the incidents-backed response is loading or unavailable. The chart reveal/glow accents are decorative and disabled when reduced-motion preferences are active. The SDLC completion rate is shown as a radial gauge and is calculated as cohort conversion from in-range triage entrants, so the rate is capped at 100% even when older tasks finish during the range.
|
||||
- **Overview** summarizes token usage/cost, autonomy, active nodes, sessions, agent runs, tasks done, model breadth, and real open signals, and includes the SDLC throughput funnel for the selected range at the bottom of the Overview content in loading, error, empty, and populated states. Its token total and Live activity snapshot token metric refresh on a bounded live cadence and animate number changes while preserving reduced-motion preferences. The sessions card uses the selected-range `ActivityAnalytics.sessions` value already loaded for the overview. The Live activity snapshot also shows the current board-state count for tasks in progress, independent of the selected analytics date range. Overview's active-agent and daily activity values count durable-agent usage events plus ephemeral task-worker execution runs, including graph-owned workflow step sessions that publish `agentRuns` lifecycle rows, with the same agent counted once per day/range if both sources record activity. Overview includes a graph-rich software-factory snapshot with the existing top-model-consumers bar, tool-category bar, top-model token-share pie, and the daily activity multi-series line chart placed before the daily activity sparkline/trend so the richer line graph sits higher in the chart grid. These reuse the already-loaded tokens, tools, activity, and signals analytics; the signals count comes from `/api/command-center/signals` and renders unavailable (`—`) while the incidents-backed response is loading or unavailable. The chart reveal/glow accents are decorative and disabled when reduced-motion preferences are active. The SDLC completion rate is shown as a radial gauge and is calculated as cohort conversion from in-range triage entrants, so the rate is capped at 100% even when older tasks finish during the range.
|
||||
<!-- FNXC:CommandCenter 2026-06-21-00:00: Command Center cost must read as an estimated, derived value from recorded token counts and the hand-maintained model pricing map; it is never persisted, and the UI must surface prices-as-of, stale low-confidence, and unavailable unknown-model states instead of implying billing truth. -->
|
||||
<!-- FNXC:CommandCenter 2026-06-22-00:00: FN-6876 requires user-maintained/LiteLLM-fetched pricing overrides to feed Tokens and Team estimates immediately without implying provider billing reconciliation. -->
|
||||
- **Tokens** breaks down token totals, estimated cost, tasks, and per-model usage. Per-model and per-provider breakdowns use the task's analytics-only actually-used model snapshot when available, so usage from settings-resolved runs appears under the real runtime model instead of `(unknown)` without changing future model resolution; estimated cost uses the same snapshot-first, legacy-fallback model identity so those resolved runs price normally when the model is in the pricing table. Estimated cost is derived at read time from recorded token counts multiplied by the effective per-model pricing table: Settings → Global Models pricing overrides win first, then the built-in fallback table is used. It is not persisted, so historical rows stay tied to current maintained prices instead of stale stored billing truth. The Tokens area shows a **prices as of** date/source for the effective table, marks pricing older than the staleness threshold as low-confidence, and shows cost unavailable for models with no pricing entry rather than guessing a price. It includes the existing token-usage-over-time chart, an additive recharts multi-series line graph, a full token-by-model bar, and a token-share pie backed by every grouped model returned by token analytics; use the granularity control to switch the time-series request between hourly, daily, and weekly buckets. The token total and charts poll on a bounded cadence, keep the previous data visible during refresh, animate decorative count/bar transitions, and disable those animations for reduced-motion users.
|
||||
- **Tools** shows autonomy ratio, tool-call volume, intervention counts, sessions, and tool categories. The area keeps the existing category bar and adds a recharts category-share pie from `ToolAnalytics.byCategory`. There is intentionally no tools line chart yet because `ToolAnalytics` does not expose a per-day tool trend; the dashboard does not fabricate one or call a new endpoint.
|
||||
- **Activity** tracks sessions, messages, active nodes, active agents, agent heartbeat runs, and stickiness. Active agents include durable-agent `usage_events` and ephemeral task-worker `agentRuns` rows in the selected range; duplicate same-day agent ids across both sources count once, while run-only task workers still make the active-agents lines and stat cards non-zero. Agent-run sheets show total, active, completed, and failed runs for the selected range, and the Agent runs/day sparkline trends runs by `agentRuns.startedAt`. The area keeps the existing live animated line charts for messages/day, active agents/day, active nodes/day, and combined throughput/day (`messages + active agents + active nodes`), and adds a recharts multi-series line graph for messages, active agents, and agent runs plus an agent-run outcome pie from the existing `agentRuns` split. These charts reuse the existing activity analytics endpoint, refresh on a bounded 15-second cadence while mounted, keep the previous data visible during refreshes, and disable decorative draw-on motion for reduced-motion users.
|
||||
- **Activity** tracks sessions, messages, active nodes, active agents, agent heartbeat runs, and stickiness. Active agents include durable-agent `usage_events` and ephemeral task-worker `agentRuns` rows in the selected range; graph-owned/new workflow step sessions publish those `agentRuns` rows as they move from active to completed/failed, duplicate same-day agent ids across both sources count once, and run-only task workers still make the active-agents lines and stat cards non-zero. Agent-run sheets show total, active, completed, and failed runs for the selected range, and the Agent runs/day sparkline trends runs by `agentRuns.startedAt`. The area keeps the existing live animated line charts for messages/day, active agents/day, active nodes/day, and combined throughput/day (`messages + active agents + active nodes`), and adds a recharts multi-series line graph for messages, active agents, and agent runs plus an agent-run outcome pie from the existing `agentRuns` split. These charts reuse the existing activity analytics endpoint, refresh on a bounded 15-second cadence while mounted, keep the previous data visible during refreshes, and disable decorative draw-on motion for reduced-motion users.
|
||||
- **Productivity** separates outcome counters (commits and pull requests), task-duration stats, and volume proxies such as modified files, lines changed, and files by language. The task-duration block counts done tasks completed in the selected range and shows average, median, p90, and total active execution time from `cumulativeActiveMs`; when no qualifying duration data exists, duration values render the unavailable `—` sentinel rather than `0`. The Lines changed card includes **Preview LOC backfill**, an explicit operator control for historical commit-association diff stats. Preview runs the project-scoped backfill in dry-run mode by default and reports scanned rows, distinct commits, updated rows, skipped unavailable commits, and skipped invalid SHAs without writing; **Apply backfill** appears after a preview and requires danger confirmation before persisting additions/deletions to `task_commit_associations`, then renders the same counts as an applied report. It keeps the files-by-language bar and adds a language-share pie from `ProductivityAnalytics.byLanguage`. There is intentionally no productivity line chart because the current productivity response has no per-day throughput or completion time series; no new endpoint is called.
|
||||
- **Team** shows the read-only agent org chart, heartbeat pause/resume backed by the existing `enginePaused` setting, a per-agent analytics table, tokens-by-agent and tasks-done-by-agent charts, and a real token-share pie from the same per-agent token totals. The org chart is styled by Command Center's Team CSS, not lazy Agents view CSS, auto-switches to a horizontal top-down tree when the container is wide enough using the same breakpoint resolver as the full Agents view, and otherwise keeps the vertical nested list inside the taller scrollable org-chart container. The org-chart scroll container supports mouse click-and-drag panning while touch devices keep native scrolling. Parent agents draw connector lines to child agents in both horizontal and vertical Team layouts across desktop and mobile breakpoints. Org nodes show only agent names so role/title description/meta text does not clutter Team operations. Metrics come only from the project-scoped `tasks` and `agents` tables: token totals and estimated cost are summed from the `tokenUsage*` columns by `assignedAgentId`, files changed counts parsed `tasks.modifiedFiles` paths, tasks done counts `column = 'done'` moves in the selected range, and in-progress / in-review values reflect current task columns. Agent name, role, and live state come from the `agents` table; deleted-agent task history falls back to the raw agent id instead of crashing. The tab uses `/api/command-center/team`, adds no schema, never calls GitHub, and intentionally leaves per-agent issues filed/fixed to FN-6653. Team has no per-day analytics series today, so it intentionally does not render a line chart or fabricate a trend. Decorative chart reveal motion uses duration tokens and is disabled for reduced-motion users.
|
||||
<!-- FNXC:CommandCenter 2026-06-27-12:00: Workflows is a read-only Command Center detail tab for custom-workflow observability; it mirrors Team metrics by workflow without adding workflow-editing controls or schema. -->
|
||||
|
||||
@@ -203,6 +203,50 @@ describe("activity-analytics", () => {
|
||||
expect(result.stickiness).toBe(1);
|
||||
});
|
||||
|
||||
it("counts workflow-step lifecycle upserts once after active to completed update", () => {
|
||||
const runId = "workflow-step-FN-7402-20260701-abcd-step-0";
|
||||
db.prepare(
|
||||
`INSERT OR IGNORE INTO agents (id, name, role, state, createdAt, updatedAt)
|
||||
VALUES (?, ?, 'executor', 'idle', ?, ?)`,
|
||||
).run("executor-FN-7402", "executor-FN-7402", "2026-03-02T12:00:00.000Z", "2026-03-02T12:00:00.000Z");
|
||||
db.prepare(
|
||||
`INSERT INTO agentRuns (id, agentId, data, startedAt, endedAt, status)
|
||||
VALUES (?, ?, ?, ?, ?, ?)`,
|
||||
).run(
|
||||
runId,
|
||||
"executor-FN-7402",
|
||||
JSON.stringify({ taskId: "FN-7402", contextSnapshot: { workflowStep: true, stepIndex: 0 } }),
|
||||
"2026-03-02T12:00:00.000Z",
|
||||
null,
|
||||
"active",
|
||||
);
|
||||
db.prepare(
|
||||
`INSERT INTO agentRuns (id, agentId, data, startedAt, endedAt, status)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
agentId = excluded.agentId,
|
||||
data = excluded.data,
|
||||
startedAt = excluded.startedAt,
|
||||
endedAt = excluded.endedAt,
|
||||
status = excluded.status`,
|
||||
).run(
|
||||
runId,
|
||||
"executor-FN-7402",
|
||||
JSON.stringify({ taskId: "FN-7402", contextSnapshot: { workflowStep: true, stepIndex: 0 }, resultJson: { success: true } }),
|
||||
"2026-03-02T12:00:00.000Z",
|
||||
"2026-03-02T12:05:00.000Z",
|
||||
"completed",
|
||||
);
|
||||
|
||||
const result = aggregateActivityAnalytics(db, { from: "2026-03-01T00:00:00.000Z", to: "2026-03-31T23:59:59.999Z" });
|
||||
|
||||
expect(result.agentRuns).toEqual({ total: 1, active: 0, completed: 1, failed: 0 });
|
||||
expect(result.activeAgents).toBe(1);
|
||||
expect(result.daily).toEqual([
|
||||
{ day: "2026-03-02", activeNodes: 0, activeAgents: 1, messages: 0, agentRuns: 1 },
|
||||
]);
|
||||
});
|
||||
|
||||
it("counts the same agent once when usage and runs occur on the same day", () => {
|
||||
emitUsageEvent(db, { kind: "tool_call", agentId: "agent-dup", nodeId: "node-1", ts: "2026-03-02T09:00:00.000Z" });
|
||||
insertAgentRun(db, { agentId: "agent-dup", startedAt: "2026-03-02T10:00:00.000Z", status: "completed" });
|
||||
|
||||
@@ -1316,6 +1316,151 @@ describe("StepSessionExecutor", () => {
|
||||
expect(prompts[1]).not.toContain("Please include the queued guidance.");
|
||||
});
|
||||
|
||||
it("publishes workflow step activity run lifecycle for dashboard analytics", async () => {
|
||||
vi.setSystemTime(new Date("2026-07-01T12:00:00.000Z"));
|
||||
const prompt = makeStepPrompt("FN-7402", 1);
|
||||
const task = makeTaskDetail({
|
||||
id: "FN-7402",
|
||||
title: "Publish workflow activity",
|
||||
lineageId: "lineage-FN-7402",
|
||||
assignedAgentId: "assigned-agent",
|
||||
prompt,
|
||||
steps: [{ name: "Implement telemetry", status: "pending" }],
|
||||
});
|
||||
const saveRun = vi.fn().mockResolvedValue(undefined);
|
||||
mockedCreateFnAgent.mockResolvedValue({ session: makeMockSession() } as any);
|
||||
|
||||
const executor = new StepSessionExecutor({
|
||||
taskDetail: task,
|
||||
worktreePath: "/project/.worktrees/main",
|
||||
rootDir: "/project",
|
||||
settings: makeSettings({ maxParallelSteps: 1 }),
|
||||
agentStore: { saveRun } as any,
|
||||
effectiveAgentId: "column-agent",
|
||||
} as any);
|
||||
|
||||
const results = await executor.executeAll();
|
||||
|
||||
expect(results).toHaveLength(1);
|
||||
expect(results[0]?.success).toBe(true);
|
||||
expect(saveRun).toHaveBeenCalledTimes(2);
|
||||
const [activeRun, completedRun] = saveRun.mock.calls.map((call) => call[0]);
|
||||
expect(activeRun).toMatchObject({
|
||||
agentId: "column-agent",
|
||||
taskId: "FN-7402",
|
||||
startedAt: "2026-07-01T12:00:00.000Z",
|
||||
endedAt: null,
|
||||
status: "active",
|
||||
invocationSource: "assignment",
|
||||
triggerDetail: "workflow-step-session",
|
||||
contextSnapshot: {
|
||||
source: "step-session-executor",
|
||||
sessionPurpose: "executor",
|
||||
workflowStep: true,
|
||||
taskId: "FN-7402",
|
||||
taskLineageId: "lineage-FN-7402",
|
||||
assignedAgentId: "assigned-agent",
|
||||
effectiveAgentId: "column-agent",
|
||||
agentId: "column-agent",
|
||||
stepIndex: 0,
|
||||
stepName: "Implement telemetry",
|
||||
},
|
||||
});
|
||||
expect(activeRun.id).toMatch(/^workflow-step-FN-7402-.*-step-0$/);
|
||||
expect(completedRun).toMatchObject({
|
||||
id: activeRun.id,
|
||||
agentId: "column-agent",
|
||||
taskId: "FN-7402",
|
||||
startedAt: activeRun.startedAt,
|
||||
endedAt: "2026-07-01T12:00:00.000Z",
|
||||
status: "completed",
|
||||
resultJson: expect.objectContaining({ success: true, retries: 0, stepIndex: 0 }),
|
||||
});
|
||||
});
|
||||
|
||||
it("publishes failed terminal workflow step activity without leaving stale active state", async () => {
|
||||
vi.setSystemTime(new Date("2026-07-01T13:00:00.000Z"));
|
||||
const prompt = makeStepPrompt("FN-7402", 1);
|
||||
const task = makeTaskDetail({
|
||||
id: "FN-7402",
|
||||
prompt,
|
||||
assignedAgentId: "assigned-agent",
|
||||
steps: [{ name: "Failing step", status: "pending" }],
|
||||
});
|
||||
const saveRun = vi.fn().mockResolvedValue(undefined);
|
||||
mockedCreateFnAgent.mockResolvedValue({ session: makeMockSession(vi.fn().mockRejectedValue(new Error("boom"))) } as any);
|
||||
|
||||
const executor = new StepSessionExecutor({
|
||||
taskDetail: task,
|
||||
worktreePath: "/project/.worktrees/main",
|
||||
rootDir: "/project",
|
||||
settings: makeSettings({ maxParallelSteps: 1 }),
|
||||
agentStore: { saveRun } as any,
|
||||
} as any);
|
||||
|
||||
const results = await executor.executeAll();
|
||||
|
||||
expect(results).toEqual([{ stepIndex: 0, success: false, error: "boom", retries: 3, tokenUsage: undefined }]);
|
||||
const terminalRun = saveRun.mock.calls.at(-1)?.[0];
|
||||
expect(saveRun).toHaveBeenCalledTimes(2);
|
||||
expect(terminalRun).toMatchObject({
|
||||
id: saveRun.mock.calls[0]?.[0].id,
|
||||
agentId: "assigned-agent",
|
||||
status: "failed",
|
||||
endedAt: expect.stringMatching(/^2026-07-01T13:00:/),
|
||||
resultJson: expect.objectContaining({ success: false, error: "boom", retries: 3 }),
|
||||
});
|
||||
expect(saveRun.mock.calls.map((call) => call[0].status)).toEqual(["active", "failed"]);
|
||||
});
|
||||
|
||||
it("uses assigned-agent and fallback executor identities for workflow activity runs", async () => {
|
||||
const prompt = makeStepPrompt("FN-7402", 1);
|
||||
const runExecutor = async (taskOverrides: Partial<TaskDetail>) => {
|
||||
const saveRun = vi.fn().mockResolvedValue(undefined);
|
||||
mockedCreateFnAgent.mockResolvedValueOnce({ session: makeMockSession() } as any);
|
||||
const executor = new StepSessionExecutor({
|
||||
taskDetail: makeTaskDetail({ id: "FN-7402", prompt, steps: [{ name: "Step 0", status: "pending" }], ...taskOverrides }),
|
||||
worktreePath: "/project/.worktrees/main",
|
||||
rootDir: "/project",
|
||||
settings: makeSettings({ maxParallelSteps: 1 }),
|
||||
agentStore: { saveRun } as any,
|
||||
} as any);
|
||||
await executor.executeAll();
|
||||
return saveRun.mock.calls[0]?.[0];
|
||||
};
|
||||
|
||||
await expect(runExecutor({ assignedAgentId: "assigned-agent" })).resolves.toMatchObject({ agentId: "assigned-agent" });
|
||||
await expect(runExecutor({ assignedAgentId: undefined })).resolves.toMatchObject({ agentId: "executor" });
|
||||
});
|
||||
|
||||
it("continues workflow execution when workflow activity publication is unavailable or failing", async () => {
|
||||
const prompt = makeStepPrompt("FN-7402", 1);
|
||||
const task = makeTaskDetail({ id: "FN-7402", prompt, steps: [{ name: "Step 0", status: "pending" }] });
|
||||
mockedCreateFnAgent.mockResolvedValue({ session: makeMockSession() } as any);
|
||||
|
||||
const withoutStore = new StepSessionExecutor({
|
||||
taskDetail: task,
|
||||
worktreePath: "/project/.worktrees/main",
|
||||
rootDir: "/project",
|
||||
settings: makeSettings({ maxParallelSteps: 1 }),
|
||||
} as any);
|
||||
await expect(withoutStore.executeAll()).resolves.toMatchObject([{ success: true }]);
|
||||
|
||||
const saveRun = vi.fn().mockRejectedValue(new Error("db offline"));
|
||||
mockedCreateFnAgent.mockResolvedValue({ session: makeMockSession() } as any);
|
||||
const withFailingStore = new StepSessionExecutor({
|
||||
taskDetail: task,
|
||||
worktreePath: "/project/.worktrees/main",
|
||||
rootDir: "/project",
|
||||
settings: makeSettings({ maxParallelSteps: 1 }),
|
||||
agentStore: { saveRun } as any,
|
||||
} as any);
|
||||
|
||||
await expect(withFailingStore.executeAll()).resolves.toMatchObject([{ success: true }]);
|
||||
expect(saveRun).toHaveBeenCalledTimes(2);
|
||||
expect(getStepSessionLogger().warn).toHaveBeenCalledWith(expect.stringContaining("Failed to publish workflow-step activity run"));
|
||||
});
|
||||
|
||||
it("happy path: 3-step task, all steps succeed", async () => {
|
||||
const prompt = makeStepPrompt("FN-001", 3);
|
||||
const task = makeTaskDetail({ prompt, steps: [
|
||||
|
||||
@@ -18,7 +18,7 @@ const execAsync = promisify(exec);
|
||||
import { existsSync } from "node:fs";
|
||||
import { rm } from "node:fs/promises";
|
||||
import type { AgentSession } from "@earendil-works/pi-coding-agent";
|
||||
import type { AgentStore, MessageStore, PermanentAgentGatingContext, ResolvedMcpServerDefinition, TaskDetail, Settings, SteeringComment, TaskStore } from "@fusion/core";
|
||||
import type { AgentHeartbeatRun, AgentStore, MessageStore, PermanentAgentGatingContext, ResolvedMcpServerDefinition, TaskDetail, Settings, SteeringComment, TaskStore } from "@fusion/core";
|
||||
import { resolvePersistAgentThinkingLog } from "@fusion/core";
|
||||
|
||||
import {
|
||||
@@ -58,6 +58,10 @@ import { activeSessionRegistry } from "./active-session-registry.js";
|
||||
|
||||
const stepExecLog = createLogger("step-session-executor");
|
||||
|
||||
type WorkflowStepActivityRun = AgentHeartbeatRun & {
|
||||
contextSnapshot: NonNullable<AgentHeartbeatRun["contextSnapshot"]>;
|
||||
};
|
||||
|
||||
// ── Exported Types ─────────────────────────────────────────────────────
|
||||
|
||||
/** Result of executing a single step. */
|
||||
@@ -1106,6 +1110,90 @@ export class StepSessionExecutor {
|
||||
};
|
||||
}
|
||||
|
||||
private createWorkflowStepActivityRun(stepIndex: number, startedAt: string): WorkflowStepActivityRun {
|
||||
const { taskDetail } = this.options;
|
||||
const step = taskDetail.steps?.[stepIndex];
|
||||
const agentId = this.options.effectiveAgentId ?? taskDetail.assignedAgentId ?? "executor";
|
||||
|
||||
/*
|
||||
* FNXC:CommandCenterActivity 2026-07-01-00:00:
|
||||
* Graph-owned workflow step sessions must publish agentRuns rows because Command Center active-agent, daily activity, and throughput metrics consume agentRuns for ephemeral task-worker activity that may not emit usage_events rows.
|
||||
*/
|
||||
return {
|
||||
id: `${generateSyntheticRunId("workflow-step", taskDetail.id)}-step-${stepIndex}`,
|
||||
agentId,
|
||||
taskId: taskDetail.id,
|
||||
startedAt,
|
||||
endedAt: null,
|
||||
status: "active",
|
||||
invocationSource: "assignment",
|
||||
triggerDetail: "workflow-step-session",
|
||||
processPid: process.pid,
|
||||
contextSnapshot: {
|
||||
source: "step-session-executor",
|
||||
sessionPurpose: "executor",
|
||||
workflowStep: true,
|
||||
taskId: taskDetail.id,
|
||||
taskLineageId: taskDetail.lineageId,
|
||||
taskTitle: taskDetail.title,
|
||||
assignedAgentId: taskDetail.assignedAgentId,
|
||||
effectiveAgentId: this.options.effectiveAgentId,
|
||||
agentId,
|
||||
stepIndex,
|
||||
stepName: step?.name ?? `Step ${stepIndex}`,
|
||||
},
|
||||
resultJson: {
|
||||
source: "step-session-executor",
|
||||
sessionPurpose: "executor",
|
||||
workflowStep: true,
|
||||
stepIndex,
|
||||
stepName: step?.name ?? `Step ${stepIndex}`,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
private async saveWorkflowStepActivityRun(run: WorkflowStepActivityRun): Promise<void> {
|
||||
const saveRun = this.options.agentStore?.saveRun?.bind(this.options.agentStore);
|
||||
if (!saveRun) return;
|
||||
|
||||
try {
|
||||
await saveRun(run);
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
stepExecLog.warn(
|
||||
`Failed to publish workflow-step activity run ${run.id} for task ${this.options.taskDetail.id}: ${msg}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private async completeWorkflowStepActivityRun(
|
||||
run: WorkflowStepActivityRun,
|
||||
status: Extract<AgentHeartbeatRun["status"], "completed" | "failed" | "terminated">,
|
||||
result: StepResult,
|
||||
): Promise<void> {
|
||||
const terminalRun: WorkflowStepActivityRun = {
|
||||
...run,
|
||||
endedAt: new Date().toISOString(),
|
||||
status,
|
||||
usageJson: result.tokenUsage
|
||||
? {
|
||||
inputTokens: result.tokenUsage.inputTokens,
|
||||
outputTokens: result.tokenUsage.outputTokens,
|
||||
cachedTokens: result.tokenUsage.cachedTokens,
|
||||
cacheWriteTokens: result.tokenUsage.cacheWriteTokens,
|
||||
}
|
||||
: run.usageJson,
|
||||
resultJson: {
|
||||
...(run.resultJson ?? {}),
|
||||
success: result.success,
|
||||
error: result.error,
|
||||
retries: result.retries,
|
||||
tokenUsage: result.tokenUsage,
|
||||
},
|
||||
};
|
||||
await this.saveWorkflowStepActivityRun(terminalRun);
|
||||
}
|
||||
|
||||
// ── Internal: Step Execution ────────────────────────────────────────
|
||||
|
||||
/**
|
||||
@@ -1143,6 +1231,9 @@ export class StepSessionExecutor {
|
||||
await semaphore.acquire();
|
||||
}
|
||||
|
||||
const activityRun = this.createWorkflowStepActivityRun(stepIndex, new Date().toISOString());
|
||||
await this.saveWorkflowStepActivityRun(activityRun);
|
||||
|
||||
const trackingKey = this.makeTrackingKey(stepIndex);
|
||||
let retries = 0;
|
||||
// Track context-limit recovery attempts separately from retry attempts.
|
||||
@@ -1152,7 +1243,9 @@ export class StepSessionExecutor {
|
||||
try {
|
||||
for (let attempt = 0; attempt <= MAX_STEP_RETRIES; attempt++) {
|
||||
if (this.aborted) {
|
||||
return { stepIndex, success: false, error: "Execution aborted", retries };
|
||||
const result: StepResult = { stepIndex, success: false, error: "Execution aborted", retries };
|
||||
await this.completeWorkflowStepActivityRun(activityRun, "terminated", result);
|
||||
return result;
|
||||
}
|
||||
|
||||
if (attempt > 0) {
|
||||
@@ -1359,6 +1452,7 @@ Follow instructions precisely and avoid unrelated changes.`,
|
||||
retries,
|
||||
tokenUsage: await this.extractStepTokenUsage(session, reusePrimarySession),
|
||||
};
|
||||
await this.completeWorkflowStepActivityRun(activityRun, "completed", result);
|
||||
this.options.onStepComplete?.(stepIndex, result);
|
||||
return result;
|
||||
} catch (err: unknown) {
|
||||
@@ -1399,6 +1493,7 @@ Follow instructions precisely and avoid unrelated changes.`,
|
||||
retries,
|
||||
tokenUsage: await this.extractStepTokenUsage(session, reusePrimarySession),
|
||||
};
|
||||
await this.completeWorkflowStepActivityRun(activityRun, "completed", result);
|
||||
this.options.onStepComplete?.(stepIndex, result);
|
||||
return result;
|
||||
} catch (reducedErr: unknown) {
|
||||
@@ -1426,6 +1521,7 @@ Follow instructions precisely and avoid unrelated changes.`,
|
||||
retries,
|
||||
tokenUsage: await this.extractStepTokenUsage(session, reusePrimarySession),
|
||||
};
|
||||
await this.completeWorkflowStepActivityRun(activityRun, "failed", result);
|
||||
this.options.onStepComplete?.(stepIndex, result);
|
||||
return result;
|
||||
}
|
||||
@@ -1457,7 +1553,9 @@ Follow instructions precisely and avoid unrelated changes.`,
|
||||
}
|
||||
|
||||
// Should not reach here, but safety fallback
|
||||
return { stepIndex, success: false, error: "Max retries exceeded", retries };
|
||||
const result: StepResult = { stepIndex, success: false, error: "Max retries exceeded", retries };
|
||||
await this.completeWorkflowStepActivityRun(activityRun, "failed", result);
|
||||
return result;
|
||||
} finally {
|
||||
// Release semaphore
|
||||
semaphore?.release();
|
||||
|
||||
Reference in New Issue
Block a user