root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

FN-8868: restore durable agent activity telemetry

Browse Source 
Restore Activity telemetry for durable agent sessions.

- Emit session-start and usage events across durable agent lanes.
- Count durable agent sessions and user messages in Activity analytics.
- Cover lifecycle wiring and telemetry persistence with tests.

Files changed: .../fn-8868-durable-agent-activity-telemetry.md    |   7 +
 docs/dashboard-guide.md                            |   8 +-
 .../message-store-user-message-telemetry.test.ts   |  73 ++++++++++
 ...mmand-center-activity-durable-agents.pg.test.ts | 158 +++++++++++++++++++++
 packages/core/src/board/activity-analytics.ts      |  12 +-
 packages/core/src/stores/message-store.ts          |  16 +++
 packages/core/src/task-store/async/async-events.ts |  11 ++
 .../dashboard/src/__tests__/chat-manager.test.ts   |  88 ++++++++++++
 packages/dashboard/src/chat.ts                     |  17 +++
 .../__tests__/agent-usage-telemetry-lanes.test.ts  |  84 +++++++++++
 .../__tests__/agent-usage-telemetry-wiring.test.ts | 119 ++++++++++++++++
 .../src/__tests__/agent-usage-telemetry.test.ts    |  33 +++++
 .../engine/src/__tests__/executor-prompt.test.ts   |   8 ++
 packages/engine/src/__tests__/merger-ai.test.ts    |  11 ++
 .../src/__tests__/merger-merge-lifecycle.test.ts   |  21 ++-
 packages/engine/src/__tests__/reviewer.test.ts     |  58 +++++++-
 .../src/__tests__/step-session-executor.test.ts    |  37 ++++-
 packages/engine/src/__tests__/triage.test.ts       |  31 ++++
 packages/engine/src/agent-heartbeat.ts             |  43 ++++++
 packages/engine/src/agents/agent-logger.ts         |  30 +++-
 .../engine/src/agents/agent-usage-telemetry.ts     |  30 ++++
 packages/engine/src/execution/reviewer.ts          |  41 +++++-
 .../engine/src/execution/step-session-executor.ts  |  12 ++
 packages/engine/src/executor.ts                    |  27 +++-
 packages/engine/src/merge/merger-ai.ts             |   7 +
 packages/engine/src/merger.ts                      | 120 +++++++++++++++-
 packages/engine/src/triage.ts                      |   5 +
 27 files changed, 1085 insertions(+), 22 deletions(-)

Fusion-Task-Id: FN-8868

Fusion-Task-Lineage: 0aa2ee4e-4d40-4adc-a510-bf8f4b1c0233

Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
2026-08-09 10:28:24 -07:00
parent 5108cfc312
commit 29bb6d0dc2
27 changed files with 1085 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/fn-8868-durable-agent-activity-telemetry.md Normal file
View File

@@ -0,0 +1,7 @@
---
"@runfusion/fusion": patch
---
summary: Restore Activity telemetry for durable agent sessions.
category: fix
dev: Durable lane session and tool events now write usage_events and Activity counts agent sessions.

8
docs/dashboard-guide.md
View File

@@ -1351,6 +1351,8 @@ Features:
<!-- 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. -->
<!-- FNXC:CommandCenterActivity 2026-08-09-10:46: Every durable and ephemeral agent lane writes one agent-session usage event, while CLI sessions remain the sole cli_sessions writer. Human chat and mailbox sends write content-free user_message events; a single-node deployment correctly reports Active nodes 0 when no mesh node id exists. -->
- **Activity telemetry** records one `usage_events` agent session for every durable and ephemeral heartbeat, executor, workflow-step, triage, reviewer, and merger lane. **Sessions** is the sum of CLI `cli_sessions` and `agent-session` `session_start` events; each session class has one writer, so they do not double-count. Human-authored chat and mailbox messages record content-free `user_message` events. **Active nodes** uses only the mesh routing node id, so it correctly remains 0 on a single-node installation with no node id.
- **Overview controls dashboard** includes AI engine stop/start backed by `globalPause` and current-project **Max concurrent tasks** plus **Max worktrees** controls. Both capacity sliders remain visible while settings load or fail, but are disabled until settings are editable; a failed load shows its error and an intentionally disabled worktree limit explains how to enable it in Settings. Max concurrency caps top-level working agents across planning, execution, and review/merge; a free project slot serves review/merge first, ready execution second, then planning, with age and task ID deciding order only within each lane. The footer reports **Waiting**, **Running (N/max)**, and **Blocked**; column headers report executing/total (live agents in the lane over card count). Nested helper agents remain parent-internal and may temporarily exceed the displayed top-level count.
<!-- FNXC:TeamArea 2026-07-18-12:30: FN-8351 moves organization export and import to the Team tab so team-level portability controls are not presented as Overview dashboard controls. -->
- **Team tab — Org export / import** lets an operator download a portable organization JSON bundle or paste one for a dry-run preview before confirming the apply step. Exports are secret-scrubbed by default: credentials and tokens are never included, while safe secret references can remain for setup in the destination project.
@@ -1361,7 +1363,7 @@ Features:
<!-- 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, task count, chat-turn count, and per-model usage. Per-model and per-provider breakdowns include task execution tokens plus supported dashboard chat, task-detail planner chat, and room responder turns when their runtime exposes authoritative session token stats; CLI-backed chat and title generation are excluded until those paths expose reliable stats. Task counts remain task-only while chat turns are counted separately, so task detail token panels stay execution-scoped and planner chat does not double-count the task it discusses. Per-model and per-provider breakdowns use the task/chat 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; 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.
- **Activity** tracks sessions, messages, active nodes, active agents, agent heartbeat runs, and stickiness. Sessions are the sum of CLI `cli_sessions` and agent-lane `usage_events` (`session_start`, `category: agent-session`), with one writer per session class so they cannot double count. Heartbeat, executor, workflow-step, triage, reviewer, and merger lanes emit those events for durable and ephemeral agents. Human chat and mailbox sends emit content-free `user_message` events. 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. Active nodes reflects mesh routing node ids only, so it is correctly zero on a single-node install. 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. -->
@@ -2423,3 +2425,7 @@ Agent creation and detail settings support a primary role plus additional role t
### Pull-request required checks
In **Settings → Merge**, pull-request mode offers **Required pull-request checks**. Enter comma-separated GitHub check names to make Fusion wait for those names independently of repository rulesets. The PR review surfaces show the same missing, pending, failed, or truncated-check-list reasons used by the merge gate. **Reset this menu** clears the setting.
### Durable agent activity telemetry
Activity includes durable and ephemeral agent sessions from heartbeat, executor, workflow-step, triage, reviewer, and merger lanes. **Sessions** is the sum of CLI session rows and `usage_events` session-start rows in the `agent-session` category; each session class has one writer. Human chat and mailbox turns supply `user_message` events. Active nodes remains zero on a single-node installation when no mesh routing node id exists.

73
packages/core/src/__tests__/message-store-user-message-telemetry.test.ts Normal file
View File

@@ -0,0 +1,73 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
const { emitUsageEvent, sendMessage } = vi.hoisted(() => ({
emitUsageEvent: vi.fn(),
sendMessage: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("../task-store/async/async-events.js", () => ({ emitUsageEvent }));
vi.mock("../async-stores/async-message-store.js", () => ({ sendMessage }));
import { MessageStore } from "../stores/message-store.js";
/**
* FNXC:CommandCenterActivity 2026-08-09-10:46:
* Mailbox telemetry is a post-persist, fail-soft side effect. These tests pin that human content
* never reaches usage metadata and a telemetry failure cannot alter the delivered message.
*/
describe("MessageStore user-message telemetry", () => {
const layer = { db: {}, projectId: "project-1" } as any;
beforeEach(() => {
vi.clearAllMocks();
sendMessage.mockResolvedValue(undefined);
emitUsageEvent.mockResolvedValue(true);
});
it("emits one content-free mailbox event for a human sender addressed to an agent", async () => {
const store = new MessageStore(null, { asyncLayer: layer });
const message = await store.sendMessage({
fromId: "user-1", fromType: "user", toId: "agent-1", toType: "agent",
content: "private mailbox text", type: "user-to-agent", metadata: { taskId: "FN-8868" },
});
expect(emitUsageEvent).toHaveBeenCalledWith(layer.db, "project-1", {
kind: "user_message", agentId: "agent-1", taskId: "FN-8868", category: "mailbox",
});
expect(JSON.stringify(emitUsageEvent.mock.calls[0])).not.toContain("private mailbox text");
expect(message.content).toBe("private mailbox text");
});
it("emits a null-agent mailbox event for a human recipient and propagates no content", async () => {
const store = new MessageStore(null, { asyncLayer: layer });
await store.sendMessage({ fromId: "user-1", fromType: "user", toId: "user-2", toType: "user", content: "human text", type: "agent-to-user" });
expect(emitUsageEvent).toHaveBeenCalledWith(layer.db, "project-1", {
kind: "user_message", agentId: null, taskId: null, category: "mailbox",
});
expect(JSON.stringify(emitUsageEvent.mock.calls[0])).not.toContain("human text");
});
it("leaves delivery successful when telemetry rejects and skips non-human senders", async () => {
emitUsageEvent.mockRejectedValueOnce(new Error("telemetry unavailable"));
const store = new MessageStore(null, { asyncLayer: layer });
await expect(store.sendMessage({
fromId: "user-1", fromType: "user", toId: "user-2", toType: "user",
content: "human text", type: "agent-to-user",
})).resolves.toMatchObject({ content: "human text" });
await Promise.resolve();
expect(emitUsageEvent).toHaveBeenCalledWith(layer.db, "project-1", {
kind: "user_message", agentId: null, taskId: null, category: "mailbox",
});
await store.sendMessage({ fromId: "agent-1", fromType: "agent", toId: "agent-2", toType: "agent", content: "agent text", type: "agent-to-agent" });
await store.sendMessage({ fromId: "system", fromType: "system", toId: "agent-2", toType: "agent", content: "system text", type: "system" });
expect(emitUsageEvent).toHaveBeenCalledTimes(1);
});
it("leaves delivery successful when telemetry throws synchronously", async () => {
emitUsageEvent.mockImplementationOnce(() => { throw new Error("telemetry unavailable"); });
const store = new MessageStore(null, { asyncLayer: layer });
await expect(store.sendMessage({ fromId: "user-1", fromType: "user", toId: "agent-1", toType: "agent", content: "delivered", type: "user-to-agent" })).resolves.toMatchObject({ content: "delivered" });
});
});

158
packages/core/src/__tests__/postgres/command-center-activity-durable-agents.pg.test.ts Normal file
View File

@@ -0,0 +1,158 @@
import { afterAll, afterEach, beforeAll, beforeEach, expect, it, vi } from "vitest";
import { sql } from "drizzle-orm";
import type { Agent, AgentHeartbeatRun, AgentStore } from "../../index.js";
import { aggregateActivityAnalytics } from "../../board/activity-analytics.js";
import * as schema from "../../postgres/schema/index.js";
import { createSharedPgTaskStoreTestHarness, pgDescribe, type SharedPgTaskStoreHarness } from "../../__test-utils__/pg-test-harness.js";
/*
* Keep the production HeartbeatMonitor in this PG test while replacing only the AI runtime.
* The runtime seam invokes the callbacks the real model runtime supplies after a tool call.
*/
vi.mock("../../../../engine/src/agents/agent-session-helpers.js", () => ({
createResolvedAgentSession: vi.fn(async (options: { onToolStart?: (name: string, args: unknown) => void; onToolEnd?: (name: string, isError: boolean, result: unknown) => void }) => {
options.onToolStart?.("Read", { path: "private-path" });
options.onToolEnd?.("Read", false, "private-result");
return { session: { dispose: vi.fn(), prompt: vi.fn(), getSessionStats: () => ({ tokens: {} }) } };
}),
extractRuntimeHint: vi.fn(),
resolveHeartbeatSessionModels: vi.fn(() => ({ defaultProvider: "durable-provider", defaultModelId: "durable-model" })),
resolveExecutorFallbackThinkingLevel: vi.fn(),
}));
vi.mock("../../../../engine/src/pi.js", () => ({ promptWithFallback: vi.fn(async () => undefined) }));
import { HeartbeatMonitor } from "../../../../engine/src/agent-heartbeat.js";
const FROM = "2026-08-09T00:00:00.000Z";
const TO = "2026-08-09T23:59:59.999Z";
const IN_RANGE = "2026-08-09T12:00:00.000Z";
/**
* FNXC:CommandCenterActivity 2026-08-09-11:29:
* A durable agent's heartbeat run and usage events share an identity but represent different
* measurements. Activity must count one active agent while preserving both run totals and the
* agent-session boundary that fixes the permanent-agent zero-telemetry symptom.
*/
pgDescribe("durable agent Activity analytics", () => {
const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ prefix: "fusion_durable_activity", projectId: "durable-project" });
beforeAll(h.beforeAll); beforeEach(h.beforeEach); afterEach(h.afterEach); afterAll(h.afterAll);
it("returns well-formed zero activity for an empty project", async () => {
const activity = await aggregateActivityAnalytics(h.layer(), { from: FROM, to: TO });
expect(activity).toMatchObject({ sessions: 0, messages: 0, activeAgents: 0 });
expect(activity.daily).toEqual([]);
});
it("counts a durable logger session and tools while de-duplicating its heartbeat agent", async () => {
const layer = h.layer();
const projectId = layer.projectId ?? "";
await layer.db.insert(schema.project.agents).values({
projectId, id: "durable-1", name: "Durable Agent", role: "executor", state: "idle",
createdAt: IN_RANGE, updatedAt: IN_RANGE,
});
await layer.db.insert(schema.project.usageEvents).values([
{ projectId, ts: IN_RANGE, kind: "session_start", agentId: "durable-1", category: "agent-session" },
{ projectId, ts: IN_RANGE, kind: "tool_call", agentId: "durable-1", toolName: "Read", category: "filesystem" },
{ projectId, ts: IN_RANGE, kind: "tool_result", agentId: "durable-1", toolName: "Read", category: "filesystem" },
{ projectId, ts: IN_RANGE, kind: "session_start", agentId: "durable-1", category: "model-router" },
]);
await layer.db.execute(sql`
INSERT INTO project.agent_runs (project_id, id, agent_id, data, started_at, status)
VALUES (${projectId}, 'durable-run-1', 'durable-1', '{}'::jsonb, ${IN_RANGE}, 'completed')
`);
const activity = await aggregateActivityAnalytics(layer, { from: FROM, to: TO });
expect(activity.sessions).toBe(1);
expect(activity.activeAgents).toBe(1);
expect(activity.agentRuns).toMatchObject({ total: 1, completed: 1 });
});
/*
FNXC:CommandCenterActivity 2026-08-09-16:48:
The durable-agent symptom must cross the production logger, canonical usage-event store, and
Activity aggregation boundary. Seeding rows alone cannot prove permanent-agent callbacks emit them.
*/
it("turns a production durable no-task heartbeat into Activity sessions and tool usage", async () => {
const layer = h.layer();
const taskStore = h.store();
/*
* FNXC:CommandCenterActivity 2026-08-09-17:16:
* The production heartbeat lane is under test; disable only optional MCP discovery so this
* isolated PG harness never attempts to open a user-level secrets directory.
*/
const heartbeatTaskStore = new Proxy(taskStore, {
get(target, property, receiver) {
if (property === "getSettingsByScope") return undefined;
return Reflect.get(target, property, receiver);
},
});
const projectId = layer.projectId ?? "";
const agent: Agent = {
id: "durable-e2e", name: "Durable E2E", role: "executor", state: "active",
soul: "Coordinate the project.", createdAt: IN_RANGE, updatedAt: IN_RANGE,
} as Agent;
const runs = new Map<string, AgentHeartbeatRun>();
const agentStore = {
getAgent: vi.fn(async () => agent),
getCachedAgent: vi.fn(() => null),
getBudgetStatus: vi.fn(async () => ({ allowed: true, budget: { period: "daily", limit: 0, used: 0, remaining: 0 } })),
startHeartbeatRun: vi.fn(async () => ({ id: "durable-e2e-run", agentId: agent.id, startedAt: IN_RANGE, endedAt: null, status: "active" })),
endHeartbeatRun: vi.fn(async (_agentId: string, runId: string, patch: Partial<AgentHeartbeatRun>) => {
runs.set(runId, { id: runId, agentId: agent.id, startedAt: IN_RANGE, endedAt: IN_RANGE, status: "completed", ...patch } as AgentHeartbeatRun);
}),
saveRun: vi.fn(async (run: AgentHeartbeatRun) => { runs.set(run.id, run); }),
getRunDetail: vi.fn(async (_agentId: string, runId: string) => runs.get(runId) ?? { id: runId, agentId: agent.id, startedAt: IN_RANGE, endedAt: IN_RANGE, status: "completed" }),
updateAgentState: vi.fn(async () => undefined),
updateAgent: vi.fn(async () => undefined),
recordHeartbeat: vi.fn(async () => undefined),
getLastBlockedState: vi.fn(async () => null),
setLastBlockedState: vi.fn(async () => undefined),
clearLastBlockedState: vi.fn(async () => undefined),
appendRunLog: vi.fn(async () => undefined),
} as unknown as AgentStore;
await layer.db.insert(schema.project.agents).values({
projectId, id: agent.id, name: agent.name, role: agent.role, state: "active",
createdAt: IN_RANGE, updatedAt: IN_RANGE,
});
const result = await new HeartbeatMonitor({
agentStore,
store: agentStore,
taskStore: heartbeatTaskStore,
rootDir: process.cwd(),
secretsStore: { listEnvExportable: vi.fn(async () => []) },
})
.executeHeartbeat({ agentId: agent.id, source: "on_demand" });
expect(result.status).toBe("completed");
const activity = await aggregateActivityAnalytics(layer, { from: "2000-01-01T00:00:00.000Z", to: "2100-01-01T00:00:00.000Z" });
expect(activity.sessions).toBeGreaterThan(0);
expect(activity.activeAgents).toBe(1);
const rows = await layer.db.execute(sql`
SELECT kind, agent_id, task_id, category FROM project.usage_events
WHERE project_id = ${projectId} ORDER BY kind
`);
expect(rows).toEqual([
{ kind: "session_start", agent_id: "durable-e2e", task_id: null, category: "agent-session" },
{ kind: "tool_call", agent_id: "durable-e2e", task_id: null, category: "read" },
{ kind: "tool_result", agent_id: "durable-e2e", task_id: null, category: "read" },
].sort((left, right) => left.kind.localeCompare(right.kind)));
});
it("sums CLI and agent sessions, honors the range, and isolates the bound project", async () => {
const layer = h.layer();
const projectId = layer.projectId ?? "";
await layer.db.insert(schema.project.usageEvents).values([
{ projectId, ts: IN_RANGE, kind: "session_start", agentId: "durable-1", category: "agent-session" },
{ projectId, ts: "2026-08-10T12:00:00.000Z", kind: "session_start", agentId: "outside", category: "agent-session" },
{ projectId: "other-project", ts: IN_RANGE, kind: "session_start", agentId: "other", category: "agent-session" },
]);
await layer.db.execute(sql`
INSERT INTO project.cli_sessions (id, purpose, project_id, adapter_id, agent_state, worktree_path, created_at, updated_at)
VALUES ('durable-cli-1', 'chat', ${projectId}, 'test', 'working', '/tmp/test', ${IN_RANGE}, ${IN_RANGE})
`);
const activity = await aggregateActivityAnalytics(layer, { from: FROM, to: TO });
expect(activity.sessions).toBe(2);
});
});

12
packages/core/src/board/activity-analytics.ts
View File

@@ -11,7 +11,8 @@ import { resolveDefaultWorkflowIr } from "../workflows/builtin-workflows.js";
* Activity analytics: distinct active nodes/agents per day, sessions, messages,
* and stickiness (DAU/MAU) over an arbitrary date range.
*
* Sessions come from `cli_sessions` (by `createdAt`); messages and node/agent
* Sessions come from `cli_sessions` plus `usage_events` session starts tagged
* `agent-session`; each session class has one writer. Messages and node/agent
* activity come from `usage_events`. Inclusivity: `from`/`to` are inclusive,
* matching `usage-events.ts`.
*
@@ -127,7 +128,7 @@ export interface SignalsAnalytics {
export interface ActivityAnalytics {
from: string | null;
to: string | null;
/** Total `session_start` events from `cli_sessions` in range. */
/** Total CLI sessions plus agent-session usage events in range. */
sessions: number;
/** Total `user_message` events in range. */
messages: number;
@@ -375,8 +376,10 @@ async function aggregatePostgresActivityAnalytics(
? sql`AND project_id = ${layer.projectId}`
: sql``;
const [sessionResult, eventSummaryResult, eventDailyResult, runStatusResult, runDailyResult, runAgentResult, monitor, funnel] = await Promise.all([
/* FNXC:ActivityAnalyticsPostgres 2026-08-09-10:29: Only agent-session rows count here; model-router session boundaries are routing telemetry, not user-visible sessions. */
const [sessionResult, agentSessionResult, eventSummaryResult, eventDailyResult, runStatusResult, runDailyResult, runAgentResult, monitor, funnel] = await Promise.all([
layer.db.execute(sql`SELECT count(*)::int AS count FROM project.cli_sessions WHERE 1=1 ${analyticsProject} ${sessionFrom} ${sessionTo}`),
layer.db.execute(sql`SELECT count(*)::int AS count FROM project.usage_events WHERE kind = 'session_start' AND category = 'agent-session' ${analyticsProject} ${eventFrom} ${eventTo}`),
layer.db.execute(sql`
SELECT
count(*) FILTER (WHERE kind = 'user_message')::int AS messages,
@@ -403,6 +406,7 @@ async function aggregatePostgresActivityAnalytics(
aggregatePostgresSdlcFunnel(layer, query),
]);
const sessionRows = sessionResult as unknown as Array<{ count?: number }>;
const agentSessionRows = agentSessionResult as unknown as Array<{ count?: number }>;
const eventSummaryRows = eventSummaryResult as unknown as PostgresEventSummaryRow[];
const eventDailyRows = eventDailyResult as unknown as PostgresEventDailyRow[];
const runStatusRows = runStatusResult as unknown as Array<{ status: string; count: number }>;
@@ -447,7 +451,7 @@ async function aggregatePostgresActivityAnalytics(
return {
from: query.from ?? null,
to: query.to ?? null,
sessions: Number(sessionRows[0]?.count ?? 0),
sessions: Number(sessionRows[0]?.count ?? 0) + Number(agentSessionRows[0]?.count ?? 0),
messages: Number(eventSummary?.messages ?? 0),
activeNodes: Number(eventSummary?.active_nodes ?? 0),
activeAgents,

16
packages/core/src/stores/message-store.ts
View File

@@ -19,6 +19,7 @@ import { DASHBOARD_USER_ID, normalizeMessageParticipant, validateMessageMetadata
import type { AsyncDataLayer } from "../postgres/data-layer.js";
import * as asyncMessageStore from "../async-stores/async-message-store.js";
import { sanitizeTextValue, sanitizeJsonbValue } from "../postgres/nul-sanitize.js";
import { emitUsageEvent } from "../task-store/async/async-events.js";
const messageStoreLog = createLogger("message-store");
@@ -224,6 +225,21 @@ export class MessageStore extends EventEmitter<MessageStoreEvents> {
this.db!.bumpLastModified();
}
/*
FNXC:CommandCenterActivity 2026-08-09-10:46:
Human mailbox sends count as activity only after durable delivery succeeds. Usage telemetry contains
routing identifiers only, never message prose, and its failure cannot delay or reverse mailbox delivery.
*/
if (this.asyncLayer && message.fromType === "user") {
try {
void emitUsageEvent(this.asyncLayer.db, this.asyncLayer.projectId ?? "", {
kind: "user_message", agentId: message.toType === "agent" ? message.toId : null,
taskId: typeof message.metadata?.taskId === "string" ? message.metadata.taskId : null,
category: "mailbox",
}).catch((error) => messageStoreLog.warn(`Failed to emit mailbox usage telemetry: ${error instanceof Error ? error.message : String(error)}`));
} catch (error) { messageStoreLog.warn(`Failed to emit mailbox usage telemetry: ${error instanceof Error ? error.message : String(error)}`); }
}
messageStoreLog.log(`MessageStore emitting message:sent id=${message.id} type=${message.type} fromId=${message.fromId} toId=${message.toId}`);
this.emit("message:sent", message);
this.emit("message:received", message);

11
packages/core/src/task-store/async/async-events.ts
View File

@@ -174,11 +174,22 @@ export async function listGoalCitations(
/**
* The set of valid usage-event kinds. Mirrors `USAGE_EVENT_KINDS`.
*/
/*
FNXC:CommandCenterActivity 2026-08-09-17:00:
The PostgreSQL writer is the production usage-event seam. It must accept every normalized
session, tool completion, and human-message kind that durable agent lanes emit; otherwise
fail-soft rejection silently recreates the Activity area's permanent-agent zero telemetry.
*/
const USAGE_EVENT_KINDS: ReadonlySet<string> = new Set([
"agent_run_started",
"agent_run_completed",
"token_usage",
"tool_call",
"tool_result",
"tool_error",
"user_message",
"session_start",
"session_stop",
"task_created",
"task_updated",
"task_moved",

88
packages/dashboard/src/__tests__/chat-manager.test.ts
View File

@@ -210,6 +210,94 @@ describe("ChatManager.sendMessage", () => {
vi.restoreAllMocks();
});
it("emits one content-free usage event after persisting a human chat turn", async () => {
const taskStore = { emitUsageEvent: vi.fn(), getSettings: vi.fn().mockResolvedValue({}) };
__setCreateResolvedAgentSession(async () => ({
session: {
prompt: vi.fn().mockResolvedValue(undefined), dispose: vi.fn(),
model: { provider: "anthropic", id: "claude-test" },
state: { messages: [{ role: "assistant", content: "done" }] },
},
}) as any);
mockChatStore.addMessage.mockImplementation((_sessionId, input) => ({
id: input.role === "user" ? "user-msg" : "assistant-msg", role: input.role,
sessionId: "chat-001", content: input.content, createdAt: "2026-08-09T00:00:00.000Z",
}));
const manager = new ChatManager(mockChatStore as any, "/tmp/test", mockAgentStore as any, undefined, undefined, undefined, taskStore as any);
await manager.sendMessage("chat-001", "private chat text");
expect(taskStore.emitUsageEvent).toHaveBeenCalledTimes(1);
expect(taskStore.emitUsageEvent).toHaveBeenCalledWith({
kind: "user_message", agentId: "agent-001", taskId: null, category: "chat",
});
expect(JSON.stringify(taskStore.emitUsageEvent.mock.calls[0])).not.toContain("private chat text");
});
it("records task-planner chat against its task without inventing an agent id", async () => {
const taskStore = { emitUsageEvent: vi.fn(), getSettings: vi.fn().mockResolvedValue({}) };
mockChatStore.getSession.mockReturnValue({ id: "chat-001", agentId: "task-planner:FN-8868", status: "active" });
__setCreateResolvedAgentSession(async () => ({
session: { prompt: vi.fn().mockResolvedValue(undefined), dispose: vi.fn(), state: { messages: [] } },
}) as any);
const manager = new ChatManager(mockChatStore as any, "/tmp/test", mockAgentStore as any, undefined, undefined, undefined, taskStore as any);
await manager.sendMessage("chat-001", "private task chat text");
expect(taskStore.emitUsageEvent).toHaveBeenCalledWith({
kind: "user_message", agentId: null, taskId: "FN-8868", category: "chat",
});
});
it("does not require a TaskStore to persist a human chat turn", async () => {
__setCreateResolvedAgentSession(async () => ({
session: { prompt: vi.fn().mockResolvedValue(undefined), dispose: vi.fn(), state: { messages: [] } },
}) as any);
await expect(createChatManager().sendMessage("chat-001", "task-store-less turn")).resolves.toBeUndefined();
});
it("keeps a rejected telemetry event out of the chat save-error path", async () => {
const taskStore = { emitUsageEvent: vi.fn().mockRejectedValue(new Error("telemetry unavailable")), getSettings: vi.fn().mockResolvedValue({}) };
const broadcast = vi.spyOn(chatStreamManager, "broadcast");
__setCreateResolvedAgentSession(async () => ({
session: { prompt: vi.fn().mockResolvedValue(undefined), dispose: vi.fn(), state: { messages: [] } },
}) as any);
await expect(new ChatManager(mockChatStore as any, "/tmp/test", mockAgentStore as any, undefined, undefined, undefined, taskStore as any)
.sendMessage("chat-001", "telemetry-failure turn")).resolves.toBeUndefined();
await Promise.resolve();
expect(taskStore.emitUsageEvent).toHaveBeenCalledTimes(1);
expect(broadcast).not.toHaveBeenCalledWith("chat-001", expect.objectContaining({ data: expect.stringContaining("Failed to save message") }), expect.anything());
});
it("emits no telemetry when the user-turn persistence fails", async () => {
const taskStore = { emitUsageEvent: vi.fn(), getSettings: vi.fn().mockResolvedValue({}) };
mockChatStore.addMessage.mockRejectedValueOnce(new Error("database unavailable"));
await expect(new ChatManager(mockChatStore as any, "/tmp/test", mockAgentStore as any, undefined, undefined, undefined, taskStore as any)
.sendMessage("chat-001", "failed user turn")).resolves.toBeUndefined();
expect(taskStore.emitUsageEvent).not.toHaveBeenCalled();
});
it("emits only the persisted user turn, never generated assistant output", async () => {
const taskStore = { emitUsageEvent: vi.fn(), getSettings: vi.fn().mockResolvedValue({}) };
__setCreateResolvedAgentSession(async () => ({
session: {
prompt: vi.fn().mockResolvedValue(undefined), dispose: vi.fn(),
state: { messages: [{ role: "assistant", content: "generated response" }] },
},
}) as any);
await new ChatManager(mockChatStore as any, "/tmp/test", mockAgentStore as any, undefined, undefined, undefined, taskStore as any)
.sendMessage("chat-001", "human turn");
expect(taskStore.emitUsageEvent).toHaveBeenCalledTimes(1);
expect(taskStore.emitUsageEvent).toHaveBeenCalledWith(expect.objectContaining({ kind: "user_message", category: "chat" }));
});
it("routes model-less QuickChat through configured default grok-cli provider", async () => {
let createOptions: any;
__setCreateResolvedAgentSession(async (options: any) => {

17
packages/dashboard/src/chat.ts
View File

@@ -2452,6 +2452,23 @@ export class ChatManager {
attachments,
});
persistedUserMessageId = persistedUserMessage.id;
/*
FNXC:CommandCenterActivity 2026-08-09-10:46:
A persisted human chat turn contributes one content-free usage event. Analytics must never enter
the message-save error path because the chat record is the user-facing source of truth.
*/
try {
// FNXC:CommandCenterActivity 2026-08-09-11:47: Task-detail planner chat
// encodes its known task identity in its synthetic agent id; retain that association
// in the content-free event rather than silently reporting it as an unscoped chat turn.
const taskId = typeof session.agentId === "string" && session.agentId.startsWith(TASK_PLANNER_CHAT_AGENT_ID_PREFIX)
? session.agentId.slice(TASK_PLANNER_CHAT_AGENT_ID_PREFIX.length).trim() || null
: null;
// FNXC:CommandCenterActivity 2026-08-09-15:18: Task-planner session ids encode a task, not a durable agent principal; never count that synthetic id as an active agent.
const agentId = taskId ? null : session.agentId ?? null;
const emitted = this.taskStore?.emitUsageEvent({ kind: "user_message", agentId, taskId, category: "chat" });
void Promise.resolve(emitted).catch(() => undefined);
} catch { /* telemetry must not enter the message-save failure path */ }
} catch (err) {
this.flushInFlightGenerationPersist(sessionId, null);
chatStreamManager.broadcast(sessionId, {

84
packages/engine/src/__tests__/agent-usage-telemetry-lanes.test.ts Normal file
View File

@@ -0,0 +1,84 @@
import { describe, expect, it, vi } from "vitest";
import type { TaskStore } from "@fusion/core";
import { AgentLogger } from "../agents/agent-logger.js";
import { HeartbeatMonitor } from "../agent-heartbeat.js";
import * as sessionHelpers from "../agents/agent-session-helpers.js";
import { attachAgentUsageTelemetry, emitAgentSessionStart, type AgentTelemetryLane } from "../agents/agent-usage-telemetry.js";
/**
* FNXC:CommandCenterActivity 2026-08-09-11:29:
* Every engine lane uses the same fail-soft logger seam. These focused fixtures exercise the
* durable/no-task identity contract without making a provider session or task-agent log a test dependency.
*/
describe("agent usage telemetry lane seam", () => {
function exercise(lane: AgentTelemetryLane, taskId: string | null, agentId = `${lane}-agent`) {
const emitUsageEvent = vi.fn().mockResolvedValue(undefined);
const appendAgentLog = vi.fn().mockResolvedValue(undefined);
const store = { emitUsageEvent, appendAgentLog } as unknown as TaskStore;
const logger = new AgentLogger({ taskId: taskId ?? undefined, appendLog: vi.fn().mockResolvedValue(undefined) });
const context = { store, lane, agentId, taskId, nodeId: taskId ? "mesh-node" : null, model: "resolved-model", provider: "resolved-provider" } as const;
attachAgentUsageTelemetry(logger, context);
emitAgentSessionStart(context);
logger.onToolStart("Read", { path: "private-path" });
logger.onToolEnd("Read", false, "private-result");
return { emitUsageEvent, appendAgentLog };
}
it.each(["heartbeat", "executor", "workflow-step", "triage", "reviewer", "merger"] as const)("emits an attributed session and tool usage for %s", (lane) => {
const { emitUsageEvent } = exercise(lane, lane === "heartbeat" ? null : "FN-8868");
expect(emitUsageEvent).toHaveBeenCalledWith(expect.objectContaining({ kind: "session_start", category: "agent-session", meta: { lane } }));
expect(emitUsageEvent).toHaveBeenCalledWith(expect.objectContaining({ kind: "tool_call", agentId: `${lane}-agent`, taskId: lane === "heartbeat" ? null : "FN-8868", model: "resolved-model" }));
expect(emitUsageEvent).toHaveBeenCalledWith(expect.objectContaining({ kind: "tool_result", provider: "resolved-provider" }));
});
it("keeps no-task heartbeat telemetry out of task agent logs and swallows persistence rejection", async () => {
const emitUsageEvent = vi.fn().mockRejectedValue(new Error("offline"));
const appendAgentLog = vi.fn();
const store = { emitUsageEvent, appendAgentLog } as unknown as TaskStore;
const logger = new AgentLogger({ appendLog: vi.fn().mockResolvedValue(undefined) });
const context = { store, lane: "heartbeat" as const, agentId: "durable-heartbeat", taskId: null };
expect(() => {
attachAgentUsageTelemetry(logger, context);
emitAgentSessionStart(context);
logger.onToolStart("Read", { path: "x" });
logger.onToolEnd("Read", false, "ok");
}).not.toThrow();
await Promise.resolve();
expect(appendAgentLog).not.toHaveBeenCalled();
});
it("emits durable no-task heartbeat usage through HeartbeatMonitor's live session callbacks", async () => {
/*
FNXC:CommandCenterActivity 2026-08-09-14:55:
The durable heartbeat regression must execute the production monitor rather than manually
invoking the seam. This proves its no-task logger receives store attachment and forwards
provider callbacks without task-log persistence.
*/
const emitUsageEvent = vi.fn().mockResolvedValue(undefined);
const appendRunLog = vi.fn().mockResolvedValue(undefined);
const heartbeatStore = {
startHeartbeatRun: vi.fn().mockResolvedValue({ id: "run-1", agentId: "durable-agent", status: "active", startedAt: new Date().toISOString() }),
getRunDetail: vi.fn(), saveRun: vi.fn(), updateAgentState: vi.fn(), updateAgent: vi.fn(), endHeartbeatRun: vi.fn(),
assignTask: vi.fn(), getBudgetStatus: vi.fn().mockResolvedValue({ isOverBudget: false, isOverThreshold: false, usagePercent: 0 }),
getCachedAgent: vi.fn().mockReturnValue(null), getLastBlockedState: vi.fn(), setLastBlockedState: vi.fn(), clearLastBlockedState: vi.fn(),
appendRunLog, getAgentsByReportsTo: vi.fn().mockResolvedValue([]), recordHeartbeat: vi.fn(),
getAgent: vi.fn().mockResolvedValue({ id: "durable-agent", name: "Durable", role: "executor", state: "active", createdAt: "", updatedAt: "", metadata: {}, soul: "patrol" }),
};
const taskStore = { getSettings: vi.fn().mockResolvedValue({}), emitUsageEvent, appendAgentLog: vi.fn(), listTasks: vi.fn().mockResolvedValue([]), selectNextTaskForAgent: vi.fn().mockResolvedValue(null) } as unknown as TaskStore;
const sessionSpy = vi.spyOn(sessionHelpers, "createResolvedAgentSession").mockImplementation(async (options: any) => {
options.onToolStart("Read", { path: "private-heartbeat-input" });
options.onToolEnd("Read", false, "private-heartbeat-output");
return { session: { prompt: vi.fn().mockResolvedValue(undefined), dispose: vi.fn(), subscribe: vi.fn() } } as any;
});
try {
await new HeartbeatMonitor({ store: heartbeatStore as any, taskStore, rootDir: "/repo" }).executeHeartbeat({ agentId: "durable-agent", source: "on_demand" });
} finally {
sessionSpy.mockRestore();
}
expect(emitUsageEvent).toHaveBeenCalledWith(expect.objectContaining({ kind: "session_start", category: "agent-session", agentId: "durable-agent", taskId: null, meta: expect.objectContaining({ lane: "heartbeat", ephemeral: false, runId: "run-1" }) }));
expect(emitUsageEvent).toHaveBeenCalledWith(expect.objectContaining({ kind: "tool_call", agentId: "durable-agent", taskId: null, toolName: "Read" }));
expect(emitUsageEvent).toHaveBeenCalledWith(expect.objectContaining({ kind: "tool_result", agentId: "durable-agent", taskId: null, toolName: "Read" }));
expect(taskStore.appendAgentLog).not.toHaveBeenCalled();
});
});

119
packages/engine/src/__tests__/agent-usage-telemetry-wiring.test.ts Normal file
View File

@@ -0,0 +1,119 @@
import { readFileSync, readdirSync } from "node:fs";
import { join, relative } from "node:path";
import { createSourceFile, forEachChild, isCallExpression, isIdentifier, isNewExpression, ScriptKind, ScriptTarget } from "typescript";
import { describe, expect, it } from "vitest";
const sourceRoot = join(process.cwd(), "src");
const WINDOW_LINES = 80;
const expectedSites = new Map([
["agent-heartbeat.ts", 2], ["executor.ts", 3], ["merger.ts", 5], ["merge/merger-ai.ts", 2],
["triage.ts", 1], ["execution/reviewer.ts", 1], ["execution/step-session-executor.ts", 1],
]);
/*
FNXC:CommandCenterActivity 2026-08-09-16:38:
Lanes that discover their model after logger construction need an initial attachment for early
callbacks and a model-refresh attachment later. Count the initial attachment separately so a
later refresh cannot mask a dark logger site in a multi-site file.
*/
const PRE_RESOLUTION_ATTACH_FILES = new Set([
"agent-heartbeat.ts", "executor.ts", "merger.ts", "triage.ts",
"execution/reviewer.ts", "execution/step-session-executor.ts",
]);
function files(directory: string): string[] {
return readdirSync(directory, { withFileTypes: true }).flatMap((entry) => {
const path = join(directory, entry.name);
if (entry.name === "__tests__") return [];
return entry.isDirectory() ? files(path) : entry.name.endsWith(".ts") ? [path] : [];
});
}
/** Parse syntax instead of regex scanning so comments and string/JSDoc literals cannot create fake sites. */
function collectSites(source: string): { loggers: number[]; attaches: number[]; preResolutionAttaches: number[]; resolvedModelAttaches: number[]; lineAt: (position: number) => number } {
const ast = createSourceFile("source.ts", source, ScriptTarget.Latest, true, ScriptKind.TS);
const loggers: number[] = [];
const attaches: number[] = [];
const preResolutionAttaches: number[] = [];
const resolvedModelAttaches: number[] = [];
const visit = (node: import("typescript").Node): void => {
if (isNewExpression(node) && isIdentifier(node.expression) && node.expression.text === "AgentLogger") loggers.push(node.getStart(ast));
if (isCallExpression(node) && isIdentifier(node.expression) && node.expression.text === "attachAgentUsageTelemetry") {
const start = node.getStart(ast);
const call = node.getText(ast);
attaches.push(start);
if (call.includes("model:")) resolvedModelAttaches.push(start);
else preResolutionAttaches.push(start);
}
forEachChild(node, visit);
};
visit(ast);
return { loggers, attaches, preResolutionAttaches, resolvedModelAttaches, lineAt: (position) => ast.getLineAndCharacterOfPosition(position).line + 1 };
}
/**
* FNXC:CommandCenterActivity 2026-08-09-10:46:
* Every executable AgentLogger construction needs its own nearby attachment, rather than a
* file-level mention that lets multi-session lanes silently lose durable telemetry.
*/
describe("FN-8868 agent usage telemetry wiring", () => {
it("pairs every executable AgentLogger construction with a local telemetry attach", () => {
const exclusions: Record<string, string> = {};
const discovered = new Map<string, number>();
let total = 0;
let paired = 0;
for (const file of files(sourceRoot)) {
const source = readFileSync(file, "utf8");
const relativeFile = relative(sourceRoot, file);
const { loggers, attaches, preResolutionAttaches, lineAt } = collectSites(source);
if (loggers.length === 0) continue;
discovered.set(relativeFile, loggers.length);
for (let index = 0; index < loggers.length; index += 1) {
const start = loggers[index];
const line = lineAt(start);
const key = `${relativeFile}:${line}`;
if (exclusions[key]) continue;
const next = loggers[index + 1] ?? source.length;
const localAttaches = PRE_RESOLUTION_ATTACH_FILES.has(relativeFile) ? preResolutionAttaches : attaches;
const isAttachedLocally = localAttaches.some((attach) => attach > start && attach < next && lineAt(attach) <= line + WINDOW_LINES);
expect(isAttachedLocally, `unpaired logger at ${key}`).toBe(true);
paired += 1;
}
expect(attaches.length, `${relativeFile} needs an attach for each logger`).toBeGreaterThanOrEqual(loggers.length);
if (PRE_RESOLUTION_ATTACH_FILES.has(relativeFile)) {
expect(preResolutionAttaches.length, `${relativeFile} needs an initial attach for each logger`).toBeGreaterThanOrEqual(loggers.length);
}
}
// A JSDoc example is absent from the TypeScript AST; do not exclude its whole file so a
// future executable construction there is discovered and must be paired like every other site.
expect(exclusions).toEqual({});
for (const [file, count] of expectedSites) {
expect(discovered.get(file), file).toBe(count);
const scanned = collectSites(readFileSync(join(sourceRoot, file), "utf8"));
// The inventory is the minimum: lanes may legitimately add model-refresh attachments.
expect(scanned.attaches.length, `${file} must attach every production logger`).toBeGreaterThanOrEqual(count);
}
/*
FNXC:CommandCenterActivity 2026-08-09-16:29:
Merger constructs five loggers before their model selection is resolved. Preserve one model-bearing
refresh per construction so removing a post-resolution attach cannot leave correctly counted but
anonymous durable telemetry behind.
*/
const merger = collectSites(readFileSync(join(sourceRoot, "merger.ts"), "utf8"));
for (let index = 0; index < merger.loggers.length; index += 1) {
const start = merger.loggers[index];
const line = merger.lineAt(start);
const next = merger.loggers[index + 1] ?? Number.POSITIVE_INFINITY;
expect(
merger.resolvedModelAttaches.some((attach) => attach > start && attach < next && merger.lineAt(attach) <= line + WINDOW_LINES),
`unpaired model-refresh attach for merger.ts:${line}`,
).toBe(true);
}
total = [...discovered.values()].reduce((sum, count) => sum + count, 0);
expect(total).toBe(15);
expect(paired).toBe(total);
});
});

33
packages/engine/src/__tests__/agent-usage-telemetry.test.ts Normal file
View File

@@ -0,0 +1,33 @@
import { describe, expect, it, vi } from "vitest";
import type { TaskStore } from "@fusion/core";
import { AgentLogger } from "../agents/agent-logger.js";
import { AGENT_SESSION_USAGE_CATEGORY, attachAgentUsageTelemetry, emitAgentSessionStart } from "../agents/agent-usage-telemetry.js";
/**
* FNXC:CommandCenterActivity 2026-08-09-11:12:
* Durable lanes attach telemetry after resolving their model, so this seam must update identity
* without making tool callbacks or session boundaries depend on telemetry persistence.
*/
describe("agent usage telemetry", () => {
it("attaches and refreshes tool identity without requiring a task log", async () => {
const emitUsageEvent = vi.fn().mockResolvedValue(undefined);
const logger = new AgentLogger({ appendLog: vi.fn().mockResolvedValue(undefined) });
const store = { emitUsageEvent } as unknown as TaskStore;
attachAgentUsageTelemetry(logger, { store, agentId: "durable-agent", taskId: null, nodeId: "node-1", model: "first", provider: "provider-a", lane: "heartbeat" });
logger.onToolStart("Bash", { command: "secret command" });
attachAgentUsageTelemetry(logger, { store, agentId: "durable-agent", taskId: null, nodeId: "node-1", model: "resolved", provider: "provider-b", lane: "heartbeat" });
logger.onToolEnd("Bash", false, "secret result");
await Promise.resolve();
expect(emitUsageEvent).toHaveBeenNthCalledWith(1, expect.objectContaining({ kind: "tool_call", taskId: null, agentId: "durable-agent", nodeId: "node-1", model: "first", provider: "provider-a" }));
expect(emitUsageEvent).toHaveBeenNthCalledWith(2, expect.objectContaining({ kind: "tool_result", taskId: null, agentId: "durable-agent", model: "resolved", provider: "provider-b" }));
});
it("emits content-free agent session boundaries fail-soft", async () => {
const emitUsageEvent = vi.fn().mockRejectedValue(new Error("offline"));
const store = { emitUsageEvent } as unknown as TaskStore;
expect(() => emitAgentSessionStart({ store, agentId: "reviewer", taskId: "FN-8868", nodeId: null, model: "validator", provider: "test", lane: "reviewer", ephemeral: false, runId: "run-1" })).not.toThrow();
expect(emitUsageEvent).toHaveBeenCalledWith(expect.objectContaining({ kind: "session_start", category: AGENT_SESSION_USAGE_CATEGORY, meta: { lane: "reviewer", ephemeral: false, runId: "run-1" } }));
expect(() => attachAgentUsageTelemetry(null, { store, lane: "reviewer" })).not.toThrow();
});
});

8
packages/engine/src/__tests__/executor-prompt.test.ts
View File

@@ -1774,6 +1774,8 @@ describe("swallowed async store failure observability", () => {
it("logs warning when sessionFile update fails during retry", async () => {
const warnSpy = vi.spyOn(executorLog, "warn");
const store = createMockStore();
const emitUsageEvent = vi.fn().mockResolvedValue(undefined);
(store as any).emitUsageEvent = emitUsageEvent;
const retrySessionFilePath = "/tmp/sessions/retry-failed.jsonl";
/*
@@ -1815,6 +1817,12 @@ describe("swallowed async store failure observability", () => {
})).resolves.toBeUndefined();
expect(mockedCreateFnAgent.mock.calls.length).toBeGreaterThanOrEqual(2);
/*
FNXC:CommandCenterActivity 2026-08-09-15:18:
A task-done-less executor retry creates a replacement runtime session. The production retry
path must publish a second boundary, rather than silently undercounting durable agent work.
*/
expect(emitUsageEvent.mock.calls.filter(([event]) => event.kind === "session_start")).toHaveLength(mockedCreateFnAgent.mock.calls.length);
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("FN-001 failed to persist retry sessionFile: retry sessionFile write failed"),
);

11
packages/engine/src/__tests__/merger-ai.test.ts
View File

@@ -101,6 +101,7 @@ function makeStore(
emit: vi.fn((event: string, payload: unknown) => { emitted.push({ event, payload }); }),
logEntry: vi.fn(async (_id: string, m: string) => { logs.push(m); }),
appendAgentLog: vi.fn(async (_id: string, m: string) => { logs.push(m); }),
emitUsageEvent: vi.fn().mockResolvedValue(undefined),
getBranchGroup: vi.fn((id: string) => (group && id === group.id ? group : null)),
recordBranchGroupMemberLanded: vi.fn(async (id: string, patch: Record<string, unknown>) => {
if (group && id === group.id) Object.assign(group, patch);
@@ -544,6 +545,16 @@ describe("runAiMerge", () => {
expect(mergerLogCalls.some(([, text, type]: [string, string, string]) => type === "tool_result" && text === "read")).toBe(true);
expect(mergerLogCalls.some(([, _text, type]: [string, string, string]) => type === "thinking")).toBe(true);
expect(mergerLogCalls.some(([, _text, type]: [string, string, string]) => type === "text")).toBe(true);
/*
FNXC:CommandCenterActivity 2026-08-09-15:35:
Exercise the live merger-ai factories, not just the shared helper, so each constructed merge and
review session retains its own usage lifecycle boundary and tool callbacks after future refactors.
*/
const usageEvents = store.emitUsageEvent.mock.calls.map(([event]: [{ kind: string; category?: string; agentId?: string | null; taskId?: string | null; toolName?: string }]) => event);
expect(usageEvents.filter((event) => event.kind === "session_start" && event.category === "agent-session")).toHaveLength(2);
expect(usageEvents.filter((event) => event.kind === "tool_call" && event.toolName === "read")).toHaveLength(2);
expect(usageEvents.every((event) => event.agentId === null && event.taskId === "FN-1")).toBe(true);
createResolvedAgentSessionMock.mockReset();
});

21
packages/engine/src/__tests__/merger-merge-lifecycle.test.ts
View File

@@ -217,6 +217,7 @@ function createMockStore(taskOverrides: Partial<Task> = {}, allTasks: Task[] = [
moveTask: vi.fn().mockResolvedValue(baseTask),
logEntry: vi.fn().mockResolvedValue(undefined),
appendAgentLog: vi.fn().mockResolvedValue(undefined),
emitUsageEvent: vi.fn().mockResolvedValue(undefined),
updateSettings: vi.fn().mockResolvedValue({}),
getSettings: vi.fn().mockResolvedValue({
...DEFAULT_SETTINGS,
@@ -1914,11 +1915,15 @@ describe("aiMergeTask — retry logic with escalating strategies", () => {
});
// Agent will be called and will fail
mockedCreateFnAgent.mockImplementation(() => {
mockedCreateFnAgent.mockImplementation((options: any) => {
agentCallCount++;
return Promise.resolve({
session: {
prompt: vi.fn().mockRejectedValue(new Error("Agent failed")),
prompt: vi.fn().mockImplementation(async () => {
options.onToolStart?.("read", { path: "src/file.ts" });
options.onToolEnd?.("read", false, "file contents");
throw new Error("Agent failed");
}),
dispose: vi.fn(),
},
} as any);
@@ -1928,6 +1933,18 @@ describe("aiMergeTask — retry logic with escalating strategies", () => {
// Should have called agent exactly once (no retries since autoResolve is disabled)
expect(agentCallCount).toBe(1);
/*
FNXC:CommandCenterActivity 2026-08-09-15:55:
A conflict-resolution session is a production merger.ts lane rather than merger-ai.ts.
Its live session boundary and tool callbacks must publish usage rows even when the merge fails.
*/
const usageEvents = (store.emitUsageEvent as ReturnType<typeof vi.fn>).mock.calls
.map(([event]: [{ kind: string; category?: string; taskId?: string | null; toolName?: string }]) => event);
expect(usageEvents).toContainEqual(expect.objectContaining({
kind: "session_start", category: "agent-session", taskId: "FN-050", meta: expect.objectContaining({ lane: "merger" }),
}));
expect(usageEvents).toContainEqual(expect.objectContaining({ kind: "tool_call", toolName: "read", taskId: "FN-050" }));
expect(usageEvents).toContainEqual(expect.objectContaining({ kind: "tool_result", toolName: "read", taskId: "FN-050" }));
});
it("attempt 2 throws when squash fails for a non-conflict reason (no U files)", async () => {

58
packages/engine/src/__tests__/reviewer.test.ts
View File

@@ -39,7 +39,7 @@ vi.mock("../pi.js", () => ({
wrapToolsWithActionGate: vi.fn((tools) => tools),
}));
import { resolveAgentPrompt } from "@fusion/core";
import { resolveAgentPrompt, type TaskStore } from "@fusion/core";
import { reviewStep, ReviewerProviderError } from "../execution/reviewer.js";
import { createFnAgent, promptWithFallback } from "../pi.js";
@@ -105,6 +105,62 @@ describe("reviewStep — model settings threading", () => {
expect(opts.defaultModelId).toBe("claude-sonnet-4-5");
});
it("emits resolved durable reviewer session and tool telemetry through the live lane callbacks", async () => {
mockedCreateFnAgent.mockImplementation(async (options) => {
options.onToolStart?.("Read", { path: "private-review-input" });
options.onToolEnd?.("Read", false, "private-review-output");
return createMockSession("### Verdict: APPROVE\n### Summary\nLooks good.");
});
const store = {
emitUsageEvent: vi.fn().mockResolvedValue(undefined),
appendAgentLog: vi.fn().mockResolvedValue(undefined),
logEntry: vi.fn().mockResolvedValue(undefined),
} as unknown as TaskStore & { emitUsageEvent: ReturnType<typeof vi.fn> };
await reviewStep("/tmp/worktree", "FN-100", 1, "Test Step", "plan", "# prompt", undefined, {
store,
taskId: "FN-100",
agentId: "durable-reviewer",
task: { assignedAgentId: "fallback-agent", effectiveNodeId: "mesh-node", nodeId: "legacy-node" },
taskValidatorProvider: "validator-provider",
taskValidatorModelId: "validator-model",
});
expect(store.emitUsageEvent).toHaveBeenCalledTimes(3);
const events = store.emitUsageEvent.mock.calls.map(([event]) => event);
const sessionStart = events.find((event) => event.kind === "session_start");
const toolCall = events.find((event) => event.kind === "tool_call");
const toolResult = events.find((event) => event.kind === "tool_result");
expect(sessionStart).toMatchObject({
kind: "session_start", category: "agent-session", taskId: "FN-100", agentId: "durable-reviewer",
nodeId: "mesh-node", model: "validator-model", provider: "validator-provider", meta: { lane: "reviewer" },
});
expect(toolCall).toMatchObject({
kind: "tool_call", taskId: "FN-100", agentId: "durable-reviewer", nodeId: "mesh-node",
model: "validator-model", provider: "validator-provider", toolName: "Read",
});
expect(toolResult).toMatchObject({
kind: "tool_result", taskId: "FN-100", agentId: "durable-reviewer", nodeId: "mesh-node",
model: "validator-model", provider: "validator-provider", toolName: "Read",
});
});
it("does not count a reviewer session when runtime construction fails", async () => {
mockedCreateFnAgent.mockRejectedValue(new Error("provider unavailable"));
const store = {
emitUsageEvent: vi.fn().mockResolvedValue(undefined),
appendAgentLog: vi.fn().mockResolvedValue(undefined),
logEntry: vi.fn().mockResolvedValue(undefined),
} as unknown as TaskStore & { emitUsageEvent: ReturnType<typeof vi.fn> };
await expect(reviewStep("/tmp/worktree", "FN-100", 1, "Test Step", "plan", "# prompt", undefined, {
store, taskId: "FN-100", agentId: "durable-reviewer",
taskValidatorProvider: "validator-provider", taskValidatorModelId: "validator-model",
})).rejects.toThrow("provider unavailable");
expect(store.emitUsageEvent).not.toHaveBeenCalledWith(expect.objectContaining({ kind: "session_start" }));
});
it("does not set model fields when ReviewOptions omits them", async () => {
mockedCreateFnAgent.mockResolvedValue(
createMockSession("### Verdict: APPROVE\n### Summary\nAll good."),

37
packages/engine/src/__tests__/step-session-executor.test.ts
View File

@@ -1214,12 +1214,15 @@ describe("StepSessionExecutor", () => {
const prompt = makeStepPrompt("FN-001", 2);
const task = makeTaskDetail({
prompt,
assignedAgentId: "durable-step-agent",
effectiveNodeId: "mesh-node-1",
steps: [
{ name: "Step 0", status: "pending" },
{ name: "Step 1", status: "pending" },
],
});
const settings = makeSettings({ maxParallelSteps: 1, runStepsInNewSessions: false });
const emitUsageEvent = vi.fn().mockResolvedValue(undefined);
let statsCall = 0;
const session = {
...makeMockSession(),
@@ -1236,7 +1239,11 @@ describe("StepSessionExecutor", () => {
};
}),
};
mockedCreateFnAgent.mockResolvedValue({ session } as any);
mockedCreateFnAgent.mockImplementationOnce(async (options: any) => {
options.onToolStart("Read", { path: "private-step-input" });
options.onToolEnd("Read", false, "private-step-output");
return { session } as any;
});
const executor = new StepSessionExecutor({
taskDetail: task,
@@ -1244,6 +1251,7 @@ describe("StepSessionExecutor", () => {
rootDir: "/project",
settings,
pluginRunner: undefined,
store: { emitUsageEvent, appendAgentLog: vi.fn().mockResolvedValue(undefined) },
} as any);
const result = await executor.executeAll();
@@ -1252,6 +1260,30 @@ describe("StepSessionExecutor", () => {
expect(result).toHaveLength(2);
expect(result.every((step) => step.success)).toBe(true);
expect(mockedCreateFnAgent).toHaveBeenCalledTimes(1);
/*
FNXC:CommandCenterActivity 2026-08-09-15:06:
Reused workflow steps share one AgentSession, so their production execution path must
publish one session_start rather than counting each prompt as a new Activity session.
*/
expect(emitUsageEvent.mock.calls.filter(([event]) => event.kind === "session_start")).toHaveLength(1);
/*
FNXC:CommandCenterActivity 2026-08-09-16:38:
Execute a real workflow-step construction path, including provider tool callbacks, so the
durable-agent telemetry regression cannot be hidden by testing the shared seam in isolation.
*/
expect(emitUsageEvent).toHaveBeenCalledWith(expect.objectContaining({
kind: "session_start", category: "agent-session", agentId: "durable-step-agent",
taskId: "FN-001", nodeId: "mesh-node-1",
meta: expect.objectContaining({ lane: "workflow-step", ephemeral: true }),
}));
expect(emitUsageEvent).toHaveBeenCalledWith(expect.objectContaining({
kind: "tool_call", toolName: "Read", agentId: "durable-step-agent",
taskId: "FN-001", nodeId: "mesh-node-1",
}));
expect(emitUsageEvent).toHaveBeenCalledWith(expect.objectContaining({
kind: "tool_result", toolName: "Read", agentId: "durable-step-agent",
taskId: "FN-001", nodeId: "mesh-node-1",
}));
expect(session.prompt).toHaveBeenCalledTimes(2);
expect(session.dispose).toHaveBeenCalledTimes(1);
expect(result[0]?.tokenUsage?.inputTokens).toBe(10);
@@ -1269,6 +1301,7 @@ describe("StepSessionExecutor", () => {
],
});
const settings = makeSettings({ maxParallelSteps: 1, runStepsInNewSessions: true });
const emitUsageEvent = vi.fn().mockResolvedValue(undefined);
const sessions = [makeMockSession(), makeMockSession()];
mockedCreateFnAgent
.mockResolvedValueOnce({ session: sessions[0] } as any)
@@ -1280,6 +1313,7 @@ describe("StepSessionExecutor", () => {
rootDir: "/project",
settings,
pluginRunner: undefined,
store: { emitUsageEvent },
} as any);
const result = await executor.executeAll();
@@ -1287,6 +1321,7 @@ describe("StepSessionExecutor", () => {
expect(result).toHaveLength(2);
expect(result.every((step) => step.success)).toBe(true);
expect(mockedCreateFnAgent).toHaveBeenCalledTimes(2);
expect(emitUsageEvent.mock.calls.filter(([event]) => event.kind === "session_start")).toHaveLength(2);
expect(sessions[0]?.prompt).toHaveBeenCalledTimes(1);
expect(sessions[1]?.prompt).toHaveBeenCalledTimes(1);
expect(sessions[0]?.dispose).toHaveBeenCalledTimes(1);

31
packages/engine/src/__tests__/triage.test.ts
View File

@@ -1080,6 +1080,35 @@ describe("fast-mode triage", () => {
expect(capturedSystemPrompt).not.toContain("## Review Level");
});
/*
FNXC:CommandCenterActivity 2026-08-09-15:35:
Triage must emit its session boundary only after the live planning runtime is constructed, so a
failed construction cannot inflate Activity while every successful durable planning session is counted.
*/
it("emits triage session telemetry through the production planning session", async () => {
const task = createTriageTask({
id: "FN-8868-TRIAGE", assignedAgentId: "durable-triage", effectiveNodeId: "mesh-node-1",
});
const emitUsageEvent = vi.fn().mockResolvedValue(undefined);
const store = createMockStore({
getTask: vi.fn().mockResolvedValue({ ...mockTaskDetail, ...task, attachments: [], comments: [] }),
emitUsageEvent,
});
mockCreateFnAgent.mockImplementationOnce(async () => ({
session: {
state: {}, sessionManager: { getLeafId: vi.fn().mockReturnValue(null) }, dispose: vi.fn(), navigateTree: vi.fn(),
prompt: vi.fn().mockResolvedValue(undefined),
},
}));
await new TriageProcessor(store, "/tmp/root").specifyTask(task);
expect(emitUsageEvent).toHaveBeenCalledWith(expect.objectContaining({
kind: "session_start", category: "agent-session", agentId: "durable-triage", taskId: task.id,
nodeId: "mesh-node-1", meta: expect.objectContaining({ lane: "triage" }),
}));
});
it("keeps standard prompt for standard tasks", async () => {
const task = createTriageTask({ id: "FN-FAST-002", executionMode: "standard" });
const store = createMockStore({
@@ -6908,6 +6937,8 @@ describe("specifyTask — status restore failure diagnostics", () => {
});
const specifyPromise = processor.specifyTask(task);
// FNXC:TriagePlanningRetry 2026-08-09-15:55: Runtime setup is async; schedule its retry sleep before advancing fake time.
await vi.advanceTimersByTimeAsync(0);
await vi.advanceTimersByTimeAsync(60_000);
await expect(specifyPromise).resolves.toBeUndefined();
expect(warnSpy).toHaveBeenCalledWith(

43
packages/engine/src/agent-heartbeat.ts
View File

@@ -44,6 +44,7 @@ import { Type, type Static } from "@earendil-works/pi-ai";
import { createHash } from "node:crypto";
import { createTaskCreateTool, createTaskLogToolWithContext, createTaskLogsReadTool, createTaskDocumentWriteTool, createTaskDocumentReadTool, createTaskReadTools, createArtifactRegisterTool, createArtifactListTool, createArtifactViewTool, createListAgentsTool, createDelegateTaskTool, createTaskAssignTool, createGetAgentConfigTool, createUpdateAgentConfigTool, createAgentCreateTool, createAgentDeleteTool, createSendMessageTool, createReadMessagesTool, createPostRoomMessageTool, createMemoryTools, createGoalRetrievalTools, createMissionTools, createIdeationTools, createReadEvaluationsTool, createUpdateIdentityTool, createReflectOnPerformanceTool, createWebFetchTool, createWorkflowListTool, createWorkflowGetTool, createWorkflowValidateTool, createWorkflowSelectTool, createTaskPromoteTool, createWorkflowCreateTool, createWorkflowUpdateTool, createWorkflowDeleteTool, createWorkflowSettingsTool, createTraitListTool, createAskQuestionTool, createResearchTools, readAgentMemoryWorkspaceLongTerm, taskCreateParams } from "./agent-tools.js";
import { AgentLogger } from "./agents/agent-logger.js";
import { attachAgentUsageTelemetry, emitAgentSessionStart } from "./agents/agent-usage-telemetry.js";
import { emitApprovalMail } from "./agents/approval-mail.js";
import {
resolveAgentInstructionsWithRatings,
@@ -2868,6 +2869,7 @@ export class HeartbeatMonitor {
persistAgentToolOutput: memorySettings?.persistAgentToolOutput,
persistAgentThinkingLog: resolvePersistAgentThinkingLog(memorySettings, { ephemeral: isAgentEphemeral }),
});
attachAgentUsageTelemetry(agentLogger, { store: taskStore, agentId, taskId: null, nodeId: null, lane: "heartbeat" });
} else if (taskId) {
agentLogger = new AgentLogger({
store: taskStore,
@@ -2877,6 +2879,7 @@ export class HeartbeatMonitor {
persistAgentToolOutput: memorySettings?.persistAgentToolOutput,
persistAgentThinkingLog: resolvePersistAgentThinkingLog(memorySettings, { ephemeral: isAgentEphemeral }),
});
attachAgentUsageTelemetry(agentLogger, { store: taskStore, agentId, taskId, nodeId: taskDetail?.effectiveNodeId ?? taskDetail?.nodeId ?? null, lane: "heartbeat" });
}
const isModelUnavailableError = (errorMessage: string): boolean => {
@@ -3064,6 +3067,19 @@ export class HeartbeatMonitor {
? await mergeEffectiveSettings(taskStore, taskDetail, heartbeatBaseSettings)
: await mergeProjectWorkflowModelLaneBaseline(taskStore, heartbeatBaseSettings);
const heartbeatSessionModels = resolveHeartbeatSessionModels(heartbeatModelSettings, agent.runtimeConfig);
// FNXC:CommandCenterActivity 2026-08-09-11:12: Heartbeat model selection happens after
// logger construction, so refresh telemetry before the session boundary and tool callbacks.
attachAgentUsageTelemetry(agentLogger, {
store: taskStore,
agentId,
taskId: taskId ?? null,
nodeId: taskDetail?.effectiveNodeId ?? taskDetail?.nodeId ?? null,
model: heartbeatSessionModels.defaultModelId ?? null,
provider: heartbeatSessionModels.defaultProvider ?? null,
lane: "heartbeat",
ephemeral: isAgentEphemeral,
runId: run.id,
});
/*
* FNXC:McpConfig 2026-06-26-00:00:
* Heartbeat runs are coding-capable agent-work sessions, so configured MCP servers must be resolved with the waking agent identity and forwarded like executor/chat lanes. Log only server counts and resolution error counts; resolved env/header contents may contain materialized secrets.
@@ -3114,6 +3130,17 @@ export class HeartbeatMonitor {
actionGateContext: this.buildActionGateContext(agent, taskId, run.id, heartbeatModelSettings?.defaultAgentPermissionPolicy),
permanentAgentGating: this.buildPermanentAgentGatingContext(agent, taskId, run.id, heartbeatModelSettings?.defaultAgentPermissionPolicy),
});
emitAgentSessionStart({
store: taskStore,
agentId,
taskId: taskId ?? null,
nodeId: taskDetail?.effectiveNodeId ?? taskDetail?.nodeId ?? null,
model: heartbeatSessionModels.defaultModelId ?? null,
provider: heartbeatSessionModels.defaultProvider ?? null,
lane: "heartbeat",
ephemeral: isAgentEphemeral,
runId: run.id,
});
/*
* FNXC:TokenAnalytics 2026-07-17-14:00:
@@ -3643,6 +3670,22 @@ export class HeartbeatMonitor {
permanentAgentGating: this.buildPermanentAgentGatingContext(agent, taskId, run.id, heartbeatModelSettings?.defaultAgentPermissionPolicy),
});
session = created.session;
/*
FNXC:CommandCenterActivity 2026-08-09-15:06:
Credential rotation constructs a replacement AgentSession, so it owns a new
boundary event rather than reusing the initial session's accounting.
*/
emitAgentSessionStart({
store: taskStore,
agentId,
taskId: taskId ?? null,
nodeId: taskDetail?.effectiveNodeId ?? taskDetail?.nodeId ?? null,
model: heartbeatSessionModels.defaultModelId ?? null,
provider: heartbeatSessionModels.defaultProvider ?? null,
lane: "heartbeat",
ephemeral: isAgentEphemeral,
runId: run.id,
});
return next;
},
} : undefined,

30
packages/engine/src/agents/agent-logger.ts
View File

@@ -213,6 +213,8 @@ export interface AgentLoggerOptions {
* node context. Omit to leave agent-log behavior unchanged.
*/
usageContext?: AgentLoggerUsageContext;
/** Store used solely for usage telemetry; supports sessions without a task log. */
usageStore?: TaskStore;
}
/**
@@ -260,6 +262,13 @@ export class AgentLogger {
private readonly persistAgentToolOutput: boolean;
private readonly persistAgentThinkingLog: boolean;
private usageContext?: AgentLoggerUsageContext;
/*
* FNXC:CommandCenterActivity 2026-08-09-10:29:
* Durable no-task heartbeat runs must publish usage telemetry, so emission cannot
* require a task id. Several lanes resolve model/provider after construction,
* therefore telemetry storage and identity are attachable after construction.
*/
private usageStore?: TaskStore;
/*
* FNXC:AgentLogging 2026-07-04-09:40:
* Task logs must expose Time To First Token once per logger/request on the first persisted visible model output. Capture the arrival time at onText/onThinking instead of flush time so buffered writes do not inflate TTFT.
@@ -291,6 +300,7 @@ export class AgentLogger {
this.persistAgentToolOutput = options.persistAgentToolOutput === true;
this.persistAgentThinkingLog = options.persistAgentThinkingLog === true;
this.usageContext = options.usageContext;
this.usageStore = options.usageStore;
// Bind callbacks so they can be passed directly as function references
this.onText = this.onText.bind(this);
@@ -305,7 +315,14 @@ export class AgentLogger {
* constructed, so it calls this once those are known.
*/
setUsageContext(context: AgentLoggerUsageContext | undefined): void {
this.usageContext = context;
this.setUsageTelemetry({ usageContext: context });
}
/** Attach or refresh telemetry independently from task-log persistence. */
setUsageTelemetry(options: { store?: TaskStore; usageContext?: AgentLoggerUsageContext }): void {
if (options.store !== undefined) this.usageStore = options.store;
/* FNXC:CommandCenterActivity 2026-08-09-16:05: Preserve setUsageContext(undefined)'s historical clearing behavior while allowing callers to refresh only the store. */
if ("usageContext" in options) this.usageContext = options.usageContext;
}
/**
@@ -323,11 +340,12 @@ export class AgentLogger {
meta?: Record<string, unknown>,
): void {
const ctx = this.usageContext;
if (!ctx || !this.store || !this.taskId) return;
const store = this.usageStore ?? this.store;
if (!ctx || !store) return;
try {
const maybePromise = this.store.emitUsageEvent({
const maybePromise = store.emitUsageEvent({
kind,
taskId: this.taskId,
taskId: this.taskId || null,
agentId: ctx.agentId ?? null,
nodeId: ctx.nodeId ?? null,
model: ctx.model ?? null,
@@ -338,10 +356,10 @@ export class AgentLogger {
});
// Swallow async rejections too so a Promise-returning store stays fail-soft.
void Promise.resolve(maybePromise).catch((err) => {
this.log.warn(`Failed to emit usage event (${kind}) for "${toolName}" on ${this.taskId}: ${err instanceof Error ? err.message : String(err)}`);
this.log.warn(`Failed to emit usage event (${kind}) for "${toolName}" on ${this.taskId || "<no task>"}: ${err instanceof Error ? err.message : String(err)}`);
});
} catch (err) {
this.log.warn(`Failed to emit usage event (${kind}) for "${toolName}" on ${this.taskId}: ${err instanceof Error ? err.message : String(err)}`);
this.log.warn(`Failed to emit usage event (${kind}) for "${toolName}" on ${this.taskId || "<no task>"}: ${err instanceof Error ? err.message : String(err)}`);
}
}

30
packages/engine/src/agents/agent-usage-telemetry.ts Normal file
View File

@@ -0,0 +1,30 @@
import type { TaskStore } from "@fusion/core";
import { AgentLogger } from "./agent-logger.js";
import { createLogger } from "../logger.js";
export const AGENT_SESSION_USAGE_CATEGORY = "agent-session";
export type AgentTelemetryLane = "heartbeat" | "executor" | "workflow-step" | "triage" | "reviewer" | "merger";
export interface AgentSessionTelemetryContext {
store: TaskStore;
agentId?: string | null;
taskId?: string | null;
nodeId?: string | null;
model?: string | null;
provider?: string | null;
lane: AgentTelemetryLane;
ephemeral?: boolean;
runId?: string;
}
const log = createLogger("agent-usage-telemetry");
/** Attach telemetry without coupling usage events to the task-log lifecycle. */
export function attachAgentUsageTelemetry(logger: AgentLogger | null | undefined, ctx: AgentSessionTelemetryContext): void {
if (!logger) return;
logger.setUsageTelemetry({ store: ctx.store, usageContext: { agentId: ctx.agentId ?? null, nodeId: ctx.nodeId ?? null, model: ctx.model ?? null, provider: ctx.provider ?? null } });
}
/** Emit a single, non-sensitive session boundary event; failures never affect an agent lane. */
export function emitAgentSessionStart(ctx: AgentSessionTelemetryContext): void {
try {
const result = ctx.store.emitUsageEvent({ kind: "session_start", taskId: ctx.taskId ?? null, agentId: ctx.agentId ?? null, nodeId: ctx.nodeId ?? null, model: ctx.model ?? null, provider: ctx.provider ?? null, category: AGENT_SESSION_USAGE_CATEGORY, meta: { lane: ctx.lane, ...(ctx.ephemeral !== undefined ? { ephemeral: ctx.ephemeral } : {}), ...(ctx.runId ? { runId: ctx.runId } : {}) } });
void Promise.resolve(result).catch((error) => log.warn(`Failed to emit agent session telemetry: ${error instanceof Error ? error.message : String(error)}`));
} catch (error) { log.warn(`Failed to emit agent session telemetry: ${error instanceof Error ? error.message : String(error)}`); }
}

41
packages/engine/src/execution/reviewer.ts
View File

@@ -34,6 +34,7 @@ import {
} from "../agents/agent-session-helpers.js";
import { buildSessionSkillContext } from "../cli-runtime/session-skill-context.js";
import { AgentLogger } from "../agents/agent-logger.js";
import { attachAgentUsageTelemetry, emitAgentSessionStart } from "../agents/agent-usage-telemetry.js";
import { reviewerLog } from "../logger.js";
import { checkSessionError } from "../errors/usage-limit-detector.js";
import {
@@ -129,8 +130,8 @@ export interface ReviewOptions {
agentId?: string;
/** Optional task title for fallback-used notification context. */
taskTitle?: string;
/** Task with optional assignedAgentId for skill selection. */
task?: { assignedAgentId?: string | null };
/** Task identity used for skill selection and usage telemetry attribution. */
task?: { assignedAgentId?: string | null; effectiveNodeId?: string | null; nodeId?: string | null };
/** User comments on the task (author === "user"). For spec reviews, the reviewer explicitly checks that every comment is addressed. */
userComments?: TaskComment[];
/** Agent prompt configuration for resolving custom reviewer prompts. */
@@ -269,7 +270,6 @@ export async function reviewStep(
persistAgentThinkingLog: resolvePersistAgentThinkingLog(effectiveSettings, { ephemeral: false }),
})
: null;
/*
FNXC:ModelResolution 2026-06-28-17:00:
Reviewer, spec-review, and workflow review-step sessions are validator-lane sessions. Resolve their primary model through the shared session helper so task reviewer overrides, project/global validator lanes, project/global defaults, and test-mode mock forcing stay identical to core model resolution instead of drifting in a reviewer-local precedence chain.
@@ -297,6 +297,26 @@ export async function reviewStep(
);
const validatorProvider = reviewerModel.provider;
const validatorModelId = reviewerModel.modelId;
/*
FNXC:CommandCenterActivity 2026-08-09-11:29:
A reviewer session exists only after its validator model is resolved. Publish its boundary and
attach tool telemetry together with the workflow principal and routing node so durable review
work is not counted as a model-less or anonymous session.
*/
if (options.store) {
const reviewerAgentId = options.agentId ?? options.task?.assignedAgentId ?? "reviewer";
const reviewerNodeId = options.task?.effectiveNodeId ?? options.task?.nodeId ?? null;
const telemetryContext = {
store: options.store,
agentId: reviewerAgentId,
taskId: options.taskId ?? null,
nodeId: reviewerNodeId,
model: validatorModelId ?? null,
provider: validatorProvider ?? null,
lane: "reviewer" as const,
};
attachAgentUsageTelemetry(agentLogger, telemetryContext);
}
const reviewerFallbackSettings: Partial<Settings> = {
...reviewerModelSettings,
@@ -543,6 +563,21 @@ export async function reviewStep(
},
});
if (options.store) {
const telemetryContext = {
store: options.store,
agentId: options.agentId ?? options.task?.assignedAgentId ?? "reviewer",
taskId: options.taskId ?? null,
nodeId: options.task?.effectiveNodeId ?? options.task?.nodeId ?? null,
model: overrides?.forceModelId ?? validatorModelId ?? null,
provider: overrides?.forceProvider ?? validatorProvider ?? null,
lane: "reviewer" as const,
};
// FNXC:CommandCenterActivity 2026-08-09-15:18: Review boundaries are recorded only after a real reviewer runtime exists, including fallback attempts.
attachAgentUsageTelemetry(agentLogger, telemetryContext);
emitAgentSessionStart(telemetryContext);
}
const reviewerModelDesc = describeModel(session);
const reviewerModelDetails = formatModelMarkerDetails(reviewerModelDesc, options.defaultThinkingLevel);
const reviewerModelMarker = `Reviewer using model: ${reviewerModelDetails}`;

12
packages/engine/src/execution/step-session-executor.ts
View File

@@ -37,6 +37,7 @@ import { installTaskWorktreeIdentityGuard } from "../worktree/worktree-hooks.js"
import { AgentSemaphore } from "../concurrency/concurrency.js";
import { StuckTaskDetector } from "../healing/stuck-task-detector.js";
import { AgentLogger, summarizeToolArgs } from "../agents/agent-logger.js";
import { attachAgentUsageTelemetry, emitAgentSessionStart } from "../agents/agent-usage-telemetry.js";
import { createLogger } from "../logger.js";
import { createFallbackModelObserver } from "../auth/fallback-model-observer.js";
import { createRunAuditor, generateSyntheticRunId } from "../util/run-audit.js";
@@ -1357,6 +1358,8 @@ export class StepSessionExecutor {
// Step-session workers are task-scoped ephemeral agents.
persistAgentThinkingLog: resolvePersistAgentThinkingLog(settings, { ephemeral: true }),
});
{ attachAgentUsageTelemetry(agentLogger, { store: this.store, agentId: taskDetail.assignedAgentId ?? null, taskId: taskDetail.id, nodeId: taskDetail.effectiveNodeId ?? taskDetail.nodeId ?? null, lane: "workflow-step" }); }
let session: AgentSession | null = null;
const localTelemetry = { agentLogger, trackingKey };
@@ -1440,6 +1443,8 @@ export class StepSessionExecutor {
this.credentialInstanceId,
);
attachAgentUsageTelemetry(agentLogger, { store: this.store, agentId: taskDetail.assignedAgentId ?? null, taskId: taskDetail.id, nodeId: taskDetail.effectiveNodeId ?? taskDetail.nodeId ?? null, model: executorModelId ?? null, provider: executorProvider ?? null, lane: "workflow-step", ephemeral: true });
if (reusePrimarySession && this.reusablePrimarySession) {
session = this.reusablePrimarySession;
} else {
@@ -1539,6 +1544,13 @@ Follow instructions precisely and avoid unrelated changes.`,
taskEnv: this.options.taskEnv,
});
session = createResult.session;
/*
FNXC:CommandCenterActivity 2026-08-09-15:06:
A reusable primary workflow session can execute multiple steps through one live
AgentSession. Record its boundary only when that session is constructed, not on each
reuse, so the Activity Sessions metric remains a session count rather than a step count.
*/
emitAgentSessionStart({ store: this.store, agentId: taskDetail.assignedAgentId ?? null, taskId: taskDetail.id, nodeId: taskDetail.effectiveNodeId ?? taskDetail.nodeId ?? null, model: executorModelId ?? null, provider: executorProvider ?? null, lane: "workflow-step", ephemeral: true });
if (reusePrimarySession) {
this.reusablePrimarySession = session;
}

27
packages/engine/src/executor.ts
View File

@@ -204,6 +204,7 @@ import {
import { BranchAttributionError, filterFilesToOwnTaskCommits } from "./execution/branch-attribution.js";
import { resolveIntegrationBranch } from "./merge/integration-branch.js";
import { AgentLogger } from "./agents/agent-logger.js";
import { attachAgentUsageTelemetry, emitAgentSessionStart } from "./agents/agent-usage-telemetry.js";
import { emitApprovalMail } from "./agents/approval-mail.js";
import { createLogger, executorLog, reviewerLog, formatError } from "./logger.js";
import { TokenCapDetector } from "./errors/token-cap-detector.js";
@@ -15494,6 +15495,8 @@ export class TaskExecutor {
}
},
});
{ attachAgentUsageTelemetry(agentLogger, { store: this.store, agentId: engineRunContext.agentId ?? null, taskId: task.id, nodeId: task.effectiveNodeId ?? task.nodeId ?? null, lane: "executor" }); }
let agentRotationEvent: import("./credential-instance-rotation.js").RotationEvent | undefined;
let agentRotationDeclined = false;
@@ -15546,13 +15549,15 @@ export class TaskExecutor {
// give the agent logger the context it needs to emit usage_events tool
// rows (KTD3). nodeId is sourced from the routed/effective node, null
// when the task has no node context.
agentLogger.setUsageContext({
attachAgentUsageTelemetry(agentLogger, {
store: this.store,
model: executorModelId ?? null,
provider: executorProvider ?? null,
nodeId: detail.effectiveNodeId ?? detail.nodeId ?? null,
agentId: engineRunContext.agentId ?? null,
taskId: task.id,
lane: "executor",
});
// Determine whether we're resuming a previous session (pause/resume)
// or starting fresh. Use file-based sessions so conversation state
// persists across pause/unpause cycles. Resume is allowed only when
@@ -15670,6 +15675,14 @@ export class TaskExecutor {
});
session = createdSession.session;
sessionFile = createdSession.sessionFile;
/*
FNXC:CommandCenterActivity 2026-08-09-15:06:
Reopening a persisted executor session after pause continues one logical AgentSession.
Emit its session boundary only for a fresh manager so resumed work cannot inflate Sessions.
*/
if (!isResuming) {
emitAgentSessionStart({ store: this.store, agentId: engineRunContext.agentId ?? null, taskId: task.id, nodeId: detail.effectiveNodeId ?? detail.nodeId ?? null, model: executorModelId ?? null, provider: executorProvider ?? null, lane: "executor" });
}
} catch (sessionStartError) {
if (await this.recoverMissingWorktreeSessionStartFailure(task, worktreePath, sessionStartError, audit)) {
return;
@@ -16106,6 +16119,8 @@ export class TaskExecutor {
taskId: task.id,
});
retrySession = createdRetrySession.session;
// FNXC:CommandCenterActivity 2026-08-09-15:18: A retry builds a distinct runtime session, so it needs its own boundary only after construction succeeds.
emitAgentSessionStart({ store: this.store, agentId: engineRunContext.agentId ?? null, taskId: task.id, nodeId: detail.effectiveNodeId ?? detail.nodeId ?? null, model: executorModelId ?? null, provider: executorProvider ?? null, lane: "executor" });
await this.captureExecutorTokenUsageBaseline(task.id, retrySession);
captureSessionTokenBaseline(retrySession);
if (createdRetrySession.sessionFile) {
@@ -19110,6 +19125,8 @@ export class TaskExecutor {
onAgentText: this.options.onAgentText,
onAgentTool: this.options.onAgentTool,
});
{ attachAgentUsageTelemetry(logger, { store: this.store, agentId: task.assignedAgentId ?? null, taskId: task.id, nodeId: task.effectiveNodeId ?? task.nodeId ?? null, lane: "executor" }); }
// Build skill selection context
let skillContext: Awaited<ReturnType<typeof buildSessionSkillContext>> | undefined;
@@ -19137,6 +19154,7 @@ export class TaskExecutor {
task.credentialInstanceId,
);
const { provider: executorProvider, modelId: executorModelId } = executorSessionModel;
attachAgentUsageTelemetry(logger, { store: this.store, agentId: task.assignedAgentId ?? null, taskId: task.id, nodeId: task.effectiveNodeId ?? task.nodeId ?? null, model: executorModelId ?? null, provider: executorProvider ?? null, lane: "executor" });
const executorFallback = resolveExecutorFallbackModel(settings);
@@ -19185,6 +19203,7 @@ Do not refactor, rename broadly, or make opportunistic improvements.
...(skillContext?.skillSelectionContext ? { skillSelection: skillContext.skillSelectionContext } : {}),
...(skillContext && skillContext.additionalSkillPaths.length > 0 ? { additionalSkillPaths: skillContext.additionalSkillPaths } : {}),
});
emitAgentSessionStart({ store: this.store, agentId: task.assignedAgentId ?? null, taskId: task.id, nodeId: task.effectiveNodeId ?? task.nodeId ?? null, model: executorModelId ?? null, provider: executorProvider ?? null, lane: "executor" });
await this.store.logEntry(
task.id,
@@ -20179,6 +20198,8 @@ You have access to the file system to review changes.${inlineFixBlock}${verdictB
this.options.onAgentTool?.(taskId, toolName, detail);
},
});
{ attachAgentUsageTelemetry(agentLogger, { store: this.store, agentId: sessionTask.assignedAgentId ?? null, taskId: task.id, nodeId: task.effectiveNodeId ?? task.nodeId ?? null, lane: "executor" }); }
// Determine primary model and an explicit fallback. Review-type workflow
// steps use the validator lane; ordinary workflow prompts use the executor
@@ -20209,6 +20230,7 @@ You have access to the file system to review changes.${inlineFixBlock}${verdictB
const primaryModelId = useOverride ? workflowStep.modelId : laneModel.modelId;
// FNXC:ProviderAuth 2026-08-01-08:39: A workflow-step model override has no paired instance selection, so only the resolved primary task lane may carry its requested credential instance. Fallback attempts must retain their provider-default behavior rather than inheriting a primary-provider identity.
const primaryCredentialInstanceId = useOverride ? undefined : laneModel.credentialInstanceId;
attachAgentUsageTelemetry(agentLogger, { store: this.store, agentId: sessionTask.assignedAgentId ?? null, taskId: task.id, nodeId: task.effectiveNodeId ?? task.nodeId ?? null, model: primaryModelId ?? null, provider: primaryProvider ?? null, lane: "executor" });
const workflowFallback = isReviewTypeWorkflowStep
? resolveValidatorFallbackModel(settings)
@@ -20409,6 +20431,7 @@ You have access to the file system to review changes.${inlineFixBlock}${verdictB
...(additionalSkillPaths ? { additionalSkillPaths } : {}),
...(readonlyCustomTools.allowed.length > 0 ? { customTools: readonlyCustomTools.allowed } : {}),
});
emitAgentSessionStart({ store: this.store, agentId: sessionTask.assignedAgentId ?? null, taskId: task.id, nodeId: task.effectiveNodeId ?? task.nodeId ?? null, model: primaryModelId ?? null, provider: primaryProvider ?? null, lane: "executor" });
const workflowModelDetails = formatModelMarkerDetails(
describeModel(session),

7
packages/engine/src/merge/merger-ai.ts
View File

@@ -69,6 +69,7 @@ import { advanceIntegrationBranchRef } from "./merger-ref-update-advance.js";
import { createResolvedAgentSession, resolveMergerSessionModel, resolveMergerThinkingLevel, resolveMergerFallbackThinkingLevel, resolveValidatorThinkingLevel } from "../agents/agent-session-helpers.js";
import { promptWithFallback } from "../pi.js";
import { AgentLogger } from "../agents/agent-logger.js";
import { attachAgentUsageTelemetry, emitAgentSessionStart } from "../agents/agent-usage-telemetry.js";
import { withRateLimitRetry } from "../errors/rate-limit-retry.js";
import { checkSessionError } from "../errors/usage-limit-detector.js";
import { accumulateSessionTokenUsage } from "../execution/session-token-usage.js";
@@ -427,6 +428,8 @@ function makeMutatingAgent(store: TaskStore, settings: Settings, taskId: string,
? (_id: string, name: string) => options.onAgentTool?.(name)
: undefined,
});
{ attachAgentUsageTelemetry(logger, { store, agentId: task?.assignedAgentId ?? null, taskId, nodeId: task?.effectiveNodeId ?? task?.nodeId ?? null, model: model.modelId ?? null, provider: model.provider ?? null, lane: "merger" }); }
const { session } = await createResolvedAgentSession({
sessionPurpose: "merger",
pluginRunner: options.pluginRunner,
@@ -450,6 +453,7 @@ function makeMutatingAgent(store: TaskStore, settings: Settings, taskId: string,
mcpServers: (await resolveMcpServersForStore(store)).servers,
taskId,
});
emitAgentSessionStart({ store, agentId: task?.assignedAgentId ?? null, taskId, nodeId: task?.effectiveNodeId ?? task?.nodeId ?? null, model: model.modelId ?? null, provider: model.provider ?? null, lane: "merger" });
options.onSession?.(session);
try {
await withRateLimitRetry(async () => {
@@ -494,6 +498,8 @@ function makeReviewAgent(store: TaskStore, settings: Settings, taskId: string, o
? (_id: string, name: string) => options.onAgentTool?.(name)
: undefined,
});
{ attachAgentUsageTelemetry(logger, { store, agentId: task?.assignedAgentId ?? null, taskId, nodeId: task?.effectiveNodeId ?? task?.nodeId ?? null, model: model.modelId ?? null, provider: model.provider ?? null, lane: "merger" }); }
const { session } = await createResolvedAgentSession({
sessionPurpose: "merger",
pluginRunner: options.pluginRunner,
@@ -520,6 +526,7 @@ function makeReviewAgent(store: TaskStore, settings: Settings, taskId: string, o
mcpServers: (await resolveMcpServersForStore(store)).servers,
taskId,
});
emitAgentSessionStart({ store, agentId: task?.assignedAgentId ?? null, taskId, nodeId: task?.effectiveNodeId ?? task?.nodeId ?? null, model: model.modelId ?? null, provider: model.provider ?? null, lane: "merger" });
options.onSession?.(session);
try {
await withRateLimitRetry(async () => {

120
packages/engine/src/merger.ts
View File

@@ -135,6 +135,7 @@ import { resolveMcpServersForStore } from "./mcp/mcp-resolution.js";
import { classifyTaskWorktree, getRegisteredWorktreeBranches, isRepoRootPath, RemovalReason, removeWorktree, type WorktreePool } from "./worktree/worktree-pool.js";
import { activeSessionRegistry } from "./agents/active-session-registry.js";
import { AgentLogger } from "./agents/agent-logger.js";
import { attachAgentUsageTelemetry, emitAgentSessionStart } from "./agents/agent-usage-telemetry.js";
import { mergerLog } from "./logger.js";
/*
@@ -1117,6 +1118,8 @@ async function attemptInMergeVerificationFix(
onAgentText: options.onAgentText,
onAgentTool: options.onAgentTool,
});
{ attachAgentUsageTelemetry(logger, { store, agentId: null, taskId, nodeId: null, lane: "merger" }); }
// Build skill selection context
let skillContext = undefined;
@@ -1148,6 +1151,18 @@ async function attemptInMergeVerificationFix(
const mergerRuntimeHint = extractRuntimeHint(assignedAgent?.runtimeConfig);
const mergerTask = await store.getTask(taskId).catch(() => undefined);
const mergerSessionModel = resolveMergerSessionModel(settings, assignedAgent?.runtimeConfig, mergerTask);
// FNXC:CommandCenterActivity 2026-08-09-11:12: Merger ownership and model resolve
// after logger construction; refresh before any model callbacks publish usage events.
attachAgentUsageTelemetry(logger, {
store,
agentId: mergerTask?.assignedAgentId ?? null,
taskId,
nodeId: mergerTask?.effectiveNodeId ?? mergerTask?.nodeId ?? null,
model: mergerSessionModel.modelId ?? null,
provider: mergerSessionModel.provider ?? null,
lane: "merger",
});
// FNXC:Settings-MergerModel 2026-07-16-00:00: merger retries use the dedicated project fallback lane before the shared global fallback pair.
@@ -1212,6 +1227,15 @@ Do not refactor, rename broadly, or make opportunistic improvements.
// Register so engine.stop() can dispose this session — without this the
// fix agent keeps streaming past shutdown because it's not the autostash
// session that the engine tracks.
emitAgentSessionStart({
store,
agentId: mergerTask?.assignedAgentId ?? null,
taskId,
nodeId: mergerTask?.effectiveNodeId ?? mergerTask?.nodeId ?? null,
model: mergerSessionModel.modelId ?? null,
provider: mergerSessionModel.provider ?? null,
lane: "merger",
});
options.onSession?.(session);
const runId = mergeRunContext?.runId;
@@ -2373,6 +2397,8 @@ async function runAiAgentForAutostashConflict(params: {
? (_id: string, name: string) => options.onAgentTool!(name)
: undefined,
});
{ attachAgentUsageTelemetry(agentLogger, { store, agentId: null, taskId, nodeId: null, lane: "merger" }); }
// Skill / runtime resolution mirrors runAiAgentForCommit.
let taskForSkillContext: Awaited<ReturnType<typeof store.getTask>> | null = null;
@@ -2401,6 +2427,18 @@ async function runAiAgentForAutostashConflict(params: {
const mergerRuntimeHint = extractRuntimeHint(assignedAgent?.runtimeConfig);
const mergerTask = await store.getTask(taskId).catch(() => undefined);
const mergerSessionModel = resolveMergerSessionModel(settings, assignedAgent?.runtimeConfig, mergerTask);
// FNXC:CommandCenterActivity 2026-08-09-11:12: Merger ownership and model resolve
// after logger construction; refresh before any model callbacks publish usage events.
attachAgentUsageTelemetry(agentLogger, {
store,
agentId: mergerTask?.assignedAgentId ?? null,
taskId,
nodeId: mergerTask?.effectiveNodeId ?? mergerTask?.nodeId ?? null,
model: mergerSessionModel.modelId ?? null,
provider: mergerSessionModel.provider ?? null,
lane: "merger",
});
// FNXC:Settings-MergerModel 2026-07-16-00:00: merger retries use the dedicated project fallback lane before the shared global fallback pair.
@@ -2477,6 +2515,15 @@ ${fileList}
taskTitle: taskForSkillContext?.title,
}),
});
emitAgentSessionStart({
store,
agentId: mergerTask?.assignedAgentId ?? null,
taskId,
nodeId: mergerTask?.effectiveNodeId ?? mergerTask?.nodeId ?? null,
model: mergerSessionModel.modelId ?? null,
provider: mergerSessionModel.provider ?? null,
lane: "merger",
});
options.onSession?.(session);
try {
@@ -2789,6 +2836,8 @@ async function runAiAgentForAutostashHardFail(params: {
? (_id: string, name: string) => options.onAgentTool!(name)
: undefined,
});
{ attachAgentUsageTelemetry(agentLogger, { store, agentId: null, taskId, nodeId: null, lane: "merger" }); }
let taskForSkillContext: Awaited<ReturnType<typeof store.getTask>> | null = null;
let skillContext = undefined;
@@ -2816,6 +2865,18 @@ async function runAiAgentForAutostashHardFail(params: {
const mergerRuntimeHint = extractRuntimeHint(assignedAgent?.runtimeConfig);
const mergerTask = await store.getTask(taskId).catch(() => undefined);
const mergerSessionModel = resolveMergerSessionModel(settings, assignedAgent?.runtimeConfig, mergerTask);
// FNXC:CommandCenterActivity 2026-08-09-11:12: Merger ownership and model resolve
// after logger construction; refresh before any model callbacks publish usage events.
attachAgentUsageTelemetry(agentLogger, {
store,
agentId: mergerTask?.assignedAgentId ?? null,
taskId,
nodeId: mergerTask?.effectiveNodeId ?? mergerTask?.nodeId ?? null,
model: mergerSessionModel.modelId ?? null,
provider: mergerSessionModel.provider ?? null,
lane: "merger",
});
// FNXC:Settings-MergerModel 2026-07-16-00:00: merger retries use the dedicated project fallback lane before the shared global fallback pair.
@@ -2902,6 +2963,15 @@ ${fileList}
taskTitle: taskForSkillContext?.title,
}),
});
emitAgentSessionStart({
store,
agentId: mergerTask?.assignedAgentId ?? null,
taskId,
nodeId: mergerTask?.effectiveNodeId ?? mergerTask?.nodeId ?? null,
model: mergerSessionModel.modelId ?? null,
provider: mergerSessionModel.provider ?? null,
lane: "merger",
});
options.onSession?.(session);
try {
@@ -5907,10 +5977,24 @@ You are assisting with a paused \`git pull --rebase\`.
? (_id, delta) => options.onAgentText?.(delta)
: undefined,
});
{ attachAgentUsageTelemetry(agentLogger, { store, agentId: null, taskId, nodeId: null, lane: "merger" }); }
throwIfAborted(options?.signal, taskId);
const mergerTask = await store.getTask(taskId).catch(() => undefined);
const mergerSessionModel = resolveMergerSessionModel(settings, options?.assignedAgentRuntimeConfig, mergerTask);
// FNXC:CommandCenterActivity 2026-08-09-11:12: Merger ownership and model resolve
// after logger construction; refresh before any model callbacks publish usage events.
attachAgentUsageTelemetry(agentLogger, {
store,
agentId: mergerTask?.assignedAgentId ?? null,
taskId,
nodeId: mergerTask?.effectiveNodeId ?? mergerTask?.nodeId ?? null,
model: mergerSessionModel.modelId ?? null,
provider: mergerSessionModel.provider ?? null,
lane: "merger",
});
// FNXC:Settings-MergerModel 2026-07-16-00:00: merger retries use the dedicated project fallback lane before the shared global fallback pair.
@@ -5953,6 +6037,15 @@ You are assisting with a paused \`git pull --rebase\`.
// Register so engine.stop() can dispose this session — without this, an
// in-progress rebase conflict resolution keeps streaming past shutdown
// (the engine only tracks the autostash session by default).
emitAgentSessionStart({
store,
agentId: mergerTask?.assignedAgentId ?? null,
taskId,
nodeId: mergerTask?.effectiveNodeId ?? mergerTask?.nodeId ?? null,
model: mergerSessionModel.modelId ?? null,
provider: mergerSessionModel.provider ?? null,
lane: "merger",
});
options?.onSession?.(session);
const prompt = [
@@ -10834,9 +10927,9 @@ async function runAiAgentForCommit(params: AiAgentParams): Promise<{ success: bo
const { message } = params as { message: string };
buildFailed = true;
buildErrorMessage = message;
return {
return {
content: [{ type: "text", text: `Build failure reported: ${message}` }],
details: undefined
details: undefined
};
},
};
@@ -10858,6 +10951,8 @@ async function runAiAgentForCommit(params: AiAgentParams): Promise<{ success: bo
? (_id, name) => options.onAgentTool!(name)
: undefined,
});
{ attachAgentUsageTelemetry(agentLogger, { store, agentId: null, taskId, nodeId: null, lane: "merger" }); }
// Resolve per-agent custom instructions for the merger role
let mergerInstructions = "";
@@ -10910,6 +11005,18 @@ async function runAiAgentForCommit(params: AiAgentParams): Promise<{ success: bo
const mergerRuntimeHint = extractRuntimeHint(assignedAgent?.runtimeConfig);
const mergerTask = await store.getTask(taskId).catch(() => undefined);
const mergerSessionModel = resolveMergerSessionModel(settings, assignedAgent?.runtimeConfig, mergerTask);
// FNXC:CommandCenterActivity 2026-08-09-11:12: Merger ownership and model resolve
// after logger construction; refresh before any model callbacks publish usage events.
attachAgentUsageTelemetry(agentLogger, {
store,
agentId: mergerTask?.assignedAgentId ?? null,
taskId,
nodeId: mergerTask?.effectiveNodeId ?? mergerTask?.nodeId ?? null,
model: mergerSessionModel.modelId ?? null,
provider: mergerSessionModel.provider ?? null,
lane: "merger",
});
// FNXC:Settings-MergerModel 2026-07-16-00:00: merger retries use the dedicated project fallback lane before the shared global fallback pair.
@@ -10961,6 +11068,15 @@ async function runAiAgentForCommit(params: AiAgentParams): Promise<{ success: bo
}),
});
emitAgentSessionStart({
store,
agentId: mergerTask?.assignedAgentId ?? null,
taskId,
nodeId: mergerTask?.effectiveNodeId ?? mergerTask?.nodeId ?? null,
model: mergerSessionModel.modelId ?? null,
provider: mergerSessionModel.provider ?? null,
lane: "merger",
});
options.onSession?.(session);
try {

5
packages/engine/src/triage.ts
View File

@@ -165,6 +165,7 @@ import {
type AgentSemaphore,
} from "./concurrency/concurrency.js";
import { AgentLogger } from "./agents/agent-logger.js";
import { attachAgentUsageTelemetry, emitAgentSessionStart } from "./agents/agent-usage-telemetry.js";
import { emitApprovalMail } from "./agents/approval-mail.js";
import { acquireActiveSessionPath, activeSessionRegistry } from "./agents/active-session-registry.js";
import {
@@ -2617,6 +2618,8 @@ export class TriageProcessor {
// for fn task logs and agent log history — no stdout spam
},
});
{ attachAgentUsageTelemetry(agentLogger, { store: this.store, agentId: task.assignedAgentId ?? "triage", taskId: task.id, nodeId: task.effectiveNodeId ?? task.nodeId ?? null, lane: "triage" }); }
// Track subtasks created during triage when breakIntoSubtasks was requested.
const createdSubtasksRef: { current: string[] } = { current: [] };
@@ -2985,6 +2988,7 @@ export class TriageProcessor {
task.planningCredentialInstanceId,
);
activePlanningProvider = planningModel.provider;
attachAgentUsageTelemetry(agentLogger, { store: this.store, agentId: task.assignedAgentId ?? "triage", taskId: task.id, nodeId: task.effectiveNodeId ?? task.nodeId ?? null, model: planningModel.modelId ?? null, provider: planningModel.provider ?? null, lane: "triage" });
const planningSessionModelOptions = {
defaultProvider: planningModel.provider,
@@ -3169,6 +3173,7 @@ export class TriageProcessor {
permanentAgentGating: this.buildPermanentAgentGatingContext(task.id, triageRunContext.runId, assignedAgent, settings.defaultAgentPermissionPolicy),
onFallbackModelUsed,
});
emitAgentSessionStart({ store: this.store, agentId: task.assignedAgentId ?? "triage", taskId: task.id, nodeId: task.effectiveNodeId ?? task.nodeId ?? null, model: planningModel.modelId ?? null, provider: planningModel.provider ?? null, lane: "triage" });
const modelDesc = formatModelMarkerDetails(describeModel(session), resolvePlanningThinkingLevel(settings, task.planningThinkingLevel ?? task.thinkingLevel));
/*