diff --git a/examples/notion-essay-pr/notion-essay-pr.smoke.test.ts b/examples/notion-essay-pr/notion-essay-pr.smoke.test.ts index 0716b067..b5dfa7d1 100644 --- a/examples/notion-essay-pr/notion-essay-pr.smoke.test.ts +++ b/examples/notion-essay-pr/notion-essay-pr.smoke.test.ts @@ -244,6 +244,13 @@ class MockNotionEssayRuntime { /* unused */ } }, + trajectory: { + async chapter() {}, + async note() {}, + async decide() {}, + async error() {}, + async done() {} + }, log: () => undefined }; } diff --git a/packages/runtime/package.json b/packages/runtime/package.json index 3135c351..5f1debef 100644 --- a/packages/runtime/package.json +++ b/packages/runtime/package.json @@ -55,6 +55,7 @@ "dependencies": { "@agent-assistant/proactive": "^0.4.32", "@agentworkforce/persona-kit": "workspace:*", - "@relayfile/adapter-core": "^0.3.29" + "@relayfile/adapter-core": "^0.3.29", + "agent-trajectories": "^0.5.3" } } diff --git a/packages/runtime/src/cloud-defaults.test.ts b/packages/runtime/src/cloud-defaults.test.ts new file mode 100644 index 00000000..4b6da271 --- /dev/null +++ b/packages/runtime/src/cloud-defaults.test.ts @@ -0,0 +1,49 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import path from 'node:path'; +import type { PersonaSpec } from '@agentworkforce/persona-kit'; +import { createCloudRuntimeDefaults } from './cloud-defaults.js'; +import type { WorkforceAgentContext, WorkforceDeploymentContext } from './types.js'; + +const persona: PersonaSpec = { + id: 'demo', + intent: 'documentation', + tags: ['documentation'], + description: 'test persona', + skills: [], + harness: 'claude', + model: 'anthropic/claude-3-5-sonnet', + systemPrompt: 'be helpful', + harnessSettings: { reasoning: 'medium', timeoutSeconds: 300 }, + cloud: true +}; + +const agent: WorkforceAgentContext = { + id: 'agent_1', + deployedName: 'docs-demo', + spawnedByAgentId: null +}; + +const deployment: WorkforceDeploymentContext = { + id: 'deployment_1', + triggerKind: 'clock', + parentDeploymentId: null +}; + +const base = { persona, agent, deployment, workspaceId: 'ws-1', log: () => {} }; + +test('cloud trajectoryRoot: explicit TRAJECTORY_ROOT wins', () => { + const defaults = createCloudRuntimeDefaults({ + ...base, + env: { TRAJECTORY_ROOT: '/custom/root', WORKFORCE_SANDBOX_ROOT: '/srv/work' } + }); + assert.equal(defaults.trajectoryRoot, '/custom/root'); +}); + +test('cloud trajectoryRoot: defaults to /.trajectories in a cloud workspace', () => { + const defaults = createCloudRuntimeDefaults({ + ...base, + env: { WORKFORCE_SANDBOX_ROOT: '/srv/work' } + }); + assert.equal(defaults.trajectoryRoot, path.join('/srv/work', '.trajectories')); +}); diff --git a/packages/runtime/src/cloud-defaults.ts b/packages/runtime/src/cloud-defaults.ts index 8c4e513c..6870ec04 100644 --- a/packages/runtime/src/cloud-defaults.ts +++ b/packages/runtime/src/cloud-defaults.ts @@ -9,6 +9,7 @@ import { resolveMcpServersLenient, resolvePersonaInputs, resolveStringMapLenient, + type AiHistMcpConfig, type PersonaSpec, type RelayMcpConfig } from '@agentworkforce/persona-kit'; @@ -62,11 +63,24 @@ export interface CloudRuntimeDefaults { workflow?: WorkflowContext; llm?: LlmContext; harnessRunner: (args: HarnessRunArgs) => Promise; + /** + * Resolved trajectory root for this deployment, or `undefined` when there is + * no cloud workspace and `TRAJECTORY_ROOT` is unset (local/tests stay opt-in, + * so the runtime never writes to a developer's cwd). The runner threads this + * into `buildCtx` as the recorder write-root; the harness spec passes the + * same value to the ai-hist MCP — that single source keeps write-root and + * MCP-root identical. + */ + trajectoryRoot?: string; } export function createCloudRuntimeDefaults(options: CloudDefaultOptions): CloudRuntimeDefaults { const env = options.env ?? process.env; const root = resolveCloudWorkspaceRoot(env); + // Single source of truth for the trajectory root, computed once. Both the + // recorder write-root (threaded into buildCtx by the runner) and the ai-hist + // MCP env (in the harness spec) consume this exact value. + const trajectoryRoot = resolveCloudTrajectoryRoot(env, root); const isSandboxOptional = options.persona.sandbox === false; const baseSandbox = createProcessSandbox(root, env); const sandbox = isSandboxOptional @@ -89,14 +103,36 @@ export function createCloudRuntimeDefaults(options: CloudDefaultOptions): CloudR files, ...(workflow ? { workflow } : {}), ...(llm ? { llm } : {}), + ...(trajectoryRoot ? { trajectoryRoot } : {}), harnessRunner: createProcessHarnessRunner({ ...options, workspaceRoot: root, - env + env, + trajectoryRoot }) }; } +/** + * Resolve the trajectory root for a cloud deployment. An explicit + * `TRAJECTORY_ROOT` always wins. Otherwise it defaults to + * `/.trajectories` — but ONLY in a real cloud workspace + * (configured mount or an accessible `/workspace`), so local/test runs stay + * opt-in via env and never write to a developer's cwd. + */ +function resolveCloudTrajectoryRoot(env: NodeJS.ProcessEnv, workspaceRoot: string): string | undefined { + const explicit = env.TRAJECTORY_ROOT?.trim(); + if (explicit) return explicit; + const hasCloudWorkspace = + firstNonEmpty( + env.WORKFORCE_SANDBOX_ROOT, + env.WORKFORCE_WORKSPACE_DIR, + env.RELAYFILE_MOUNT_ROOT, + env.RELAYFILE_ROOT + ) !== undefined || canAccessSync('/workspace'); + return hasCloudWorkspace ? path.join(workspaceRoot, '.trajectories') : undefined; +} + function resolveCloudWorkspaceRoot(env: NodeJS.ProcessEnv): string { const configured = firstNonEmpty( env.WORKFORCE_SANDBOX_ROOT, @@ -413,6 +449,7 @@ function delay(ms: number): Promise { function createProcessHarnessRunner(args: CloudDefaultOptions & { workspaceRoot: string; env: NodeJS.ProcessEnv; + trajectoryRoot?: string; }): (run: HarnessRunArgs) => Promise { return async (run) => { // harness/model/systemPrompt are optional on the persona spec (pure @@ -461,6 +498,14 @@ function createProcessHarnessRunner(args: CloudDefaultOptions & { }); const task = run.prompt; const relayMcp = resolveRelayMcpFromEnv(args.env); + // Inject the ai-hist MCP (the "why" + "how" retrieval surface) by default, + // unless the persona disabled trajectory recording. resolveAiHistFromEnv + // keys off the SAME TRAJECTORY_ROOT the runtime recorder writes to, so the + // MCP reads back exactly what this deployment wrote. + const aiHist = + args.persona.recordTrajectories === false + ? undefined + : resolveAiHistFromEnv(args.env, args.trajectoryRoot); const specInput = { harness, personaId: args.persona.id, @@ -471,7 +516,8 @@ function createProcessHarnessRunner(args: CloudDefaultOptions & { permissions: args.persona.permissions, task, name: args.persona.id, - workingDirectory: cwd + workingDirectory: cwd, + ...(aiHist ? { aiHist } : {}) }; const brokerRelayHarness = relayMcp && (harness === 'claude' || harness === 'codex'); let spec = buildNonInteractiveSpec({ @@ -599,6 +645,30 @@ function resolveRelayMcpFromEnv(env: NodeJS.ProcessEnv): RelayMcpConfig | undefi }; } +/** + * Resolve the ai-hist MCP config from env. Mirrors the CLI helper so cloud and + * local spawns inject the same server. `WORKFORCE_AIHIST_DISABLED` (`1`/`true`) + * is the escape hatch. Both fields are optional — when `TRAJECTORY_ROOT` is + * unset the MCP falls back to its own discovery, which matches the runtime + * recorder (also keyed off `TRAJECTORY_ROOT`): with no root, the recorder + * writes nothing, so there is nothing to mis-read. + */ +function resolveAiHistFromEnv( + env: NodeJS.ProcessEnv, + defaultTrajectoryRoot?: string +): AiHistMcpConfig | undefined { + const disabled = env.WORKFORCE_AIHIST_DISABLED?.trim(); + if (disabled === '1' || disabled === 'true') return undefined; + // env.TRAJECTORY_ROOT wins; otherwise the deployment default (same value the + // recorder writes to) — keeps the MCP read-root identical to the write-root. + const trajectoryRoot = env.TRAJECTORY_ROOT?.trim() || defaultTrajectoryRoot; + const dbPath = env.AI_HIST_DB?.trim(); + return { + ...(trajectoryRoot ? { trajectoryRoot } : {}), + ...(dbPath ? { dbPath } : {}) + }; +} + async function resolveAgentRelayBrokerMcpArgs(args: { cli: 'claude' | 'codex'; env: NodeJS.ProcessEnv; diff --git a/packages/runtime/src/ctx.ts b/packages/runtime/src/ctx.ts index 737ca29b..1289169f 100644 --- a/packages/runtime/src/ctx.ts +++ b/packages/runtime/src/ctx.ts @@ -13,6 +13,7 @@ import type { WorkforceDeploymentContext, WorkflowContext } from './types.js'; +import { attachTrajectoryRecorder, createTrajectoryRecorder } from './trajectory.js'; type AgentInputValue = string | number | boolean | null | undefined; @@ -55,6 +56,14 @@ export interface CtxBuildOptions { integrations?: Record; log?: WorkforceCtx['log']; harnessRunner: WorkforceCtx['harness']['run']; + /** + * Root directory for per-run trajectory contract files. Defaults to + * `env.TRAJECTORY_ROOT`; when neither resolves, trajectory recording is + * disabled (the runtime never writes to the process cwd). The cloud runtime + * sets this to the same value it passes to the injected ai-hist MCP so the + * MCP reads back exactly what the runtime wrote. + */ + trajectoryRoot?: string; } const NOOP_MEMORY: MemoryContext = { @@ -126,6 +135,17 @@ export function buildCtx(options: CtxBuildOptions): WorkforceCtx { }; const log = options.log ?? defaultLog; const files = options.files ?? filesFromSandbox(options.sandbox); + const agentName = options.agentName ?? options.persona.id; + // Per-persona trajectory recorder (the WHY). No-op unless recording is on + // (`recordTrajectories !== false`) and a trajectory root resolves. + const trajectoryRecorder = createTrajectoryRecorder({ + personaId: options.persona.id, + agentName, + workspaceId: options.workspaceId, + recordTrajectories: options.persona.recordTrajectories, + ...(options.trajectoryRoot ? { trajectoryRoot: options.trajectoryRoot } : {}), + log + }); const ctx: WorkforceCtx = { persona: buildPersonaContext(options.persona, mergedAgentInputValues), agent: { @@ -135,7 +155,7 @@ export function buildCtx(options: CtxBuildOptions): WorkforceCtx { }, deployment, workspaceId: options.workspaceId, - agentName: options.agentName ?? options.persona.id, + agentName, llm: options.llm ?? UNAVAILABLE_LLM, harness: { run: options.harnessRunner }, sandbox: options.sandbox, @@ -144,8 +164,12 @@ export function buildCtx(options: CtxBuildOptions): WorkforceCtx { memory: options.memory ?? defaultMemoryFor(options.persona.memory, options.workspaceId, log), workflow: options.workflow ?? UNAVAILABLE_WORKFLOW, schedule: options.schedule ?? UNAVAILABLE_SCHEDULE, + trajectory: trajectoryRecorder.context, log }; + // The runner drives the recorder's lifecycle (begin/complete/fail) via this + // non-enumerable handle, keeping those internals off the public ctx surface. + attachTrajectoryRecorder(ctx, trajectoryRecorder); // Optional per-integration subsystems attach as named ctx fields. The // cloud-default runtime does not populate this — handlers read provider @@ -186,6 +210,7 @@ const CORE_CTX_FIELDS: ReadonlySet = new Set([ 'memory', 'workflow', 'schedule', + 'trajectory', 'log' ]); diff --git a/packages/runtime/src/runner.ts b/packages/runtime/src/runner.ts index 811c111c..100aa588 100644 --- a/packages/runtime/src/runner.ts +++ b/packages/runtime/src/runner.ts @@ -1,6 +1,7 @@ import type { AgentSpec, PersonaSpec } from '@agentworkforce/persona-kit'; import { createCloudRuntimeDefaults } from './cloud-defaults.js'; import { buildCtx, type CtxBuildOptions } from './ctx.js'; +import { getTrajectoryRecorder, type TrajectoryRecorder } from './trajectory.js'; import { isWorkforceHandler } from './handler.js'; import { shimEnvelope, type RawGatewayEnvelope } from './shim.js'; import type { @@ -123,6 +124,10 @@ export async function startRunner(options: StartRunnerOptions): Promise { : {}), ...(options.subsystems?.schedule ? { schedule: options.subsystems.schedule } : {}), ...(options.subsystems?.log ? { log: options.subsystems.log } : {}), + // Recorder write-root resolved once by cloud-defaults; identical to the + // value cloud-defaults passes to the ai-hist MCP. Undefined locally/in + // tests (recording stays opt-in via TRAJECTORY_ROOT). + ...(cloudDefaults.trajectoryRoot ? { trajectoryRoot: cloudDefaults.trajectoryRoot } : {}), ...(Object.keys(integrations).length > 0 ? { integrations } : {}) }); @@ -134,6 +139,7 @@ export async function startRunner(options: StartRunnerOptions): Promise { integrations: options.persona.integrations ? Object.keys(options.persona.integrations) : [] }); + const recorder = getTrajectoryRecorder(ctx); const stream = options.envelopes ?? readEnvelopesFromStdin(); for await (const raw of stream) { const event = shimEnvelope(raw); @@ -141,7 +147,7 @@ export async function startRunner(options: StartRunnerOptions): Promise { ctx.log('warn', 'runner.envelope.unsupported', { rawId: raw.id, rawType: raw.type }); continue; } - await dispatch(ctx, handlerFn, event); + await dispatch(ctx, handlerFn, event, recorder); } ctx.log('info', 'runner.envelope-stream.ended', { persona: options.persona.id }); @@ -155,9 +161,13 @@ function defaultRunnerLog(level: 'debug' | 'info' | 'warn' | 'error', message: s async function dispatch( ctx: Parameters[0], fn: WorkforceHandler, - event: WorkforceEvent + event: WorkforceEvent, + recorder: TrajectoryRecorder ): Promise { const t0 = Date.now(); + // Open a trajectory for this run. The handler narrates via ctx.trajectory.*; + // begin/complete/fail never throw (recording is best-effort observability). + await recorder.begin(event); try { await fn(ctx, event); ctx.log('info', 'runner.handler.ok', { @@ -166,6 +176,8 @@ async function dispatch( type: event.source === 'cron' ? 'cron.tick' : event.type, durationMs: Date.now() - t0 }); + // Auto-finalize (no-op if the handler already called ctx.trajectory.done). + await recorder.complete(); } catch (err) { ctx.log('error', 'runner.handler.error', { eventId: event.id, @@ -176,6 +188,7 @@ async function dispatch( error: err instanceof Error ? err.message : String(err), stack: err instanceof Error ? err.stack : undefined }); + await recorder.fail(err); // Surface the failure to the outer process so the deploy layer can // retry. Throwing here would tear down the for-await loop; the deploy // layer reads the structured log line above instead. diff --git a/packages/runtime/src/trajectory.test.ts b/packages/runtime/src/trajectory.test.ts new file mode 100644 index 00000000..6323df76 --- /dev/null +++ b/packages/runtime/src/trajectory.test.ts @@ -0,0 +1,187 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtemp, readFile, readdir, rm } from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; +import { createTrajectoryRecorder } from './trajectory.js'; +import type { CompactedTrajectoryContract, WorkforceEvent } from './types.js'; + +const cronEvent: WorkforceEvent = { + source: 'cron', + id: 'evt-1', + occurredAt: '2026-01-01T00:00:00.000Z', + attempt: 1, + workspaceId: 'ws-1', + name: 'weekly', + cron: '0 9 * * 6' +}; + +const silentLog = () => {}; + +async function withRoot(fn: (root: string) => Promise): Promise { + const root = await mkdtemp(path.join(os.tmpdir(), 'wf-traj-')); + try { + await fn(root); + } finally { + await rm(root, { recursive: true, force: true }); + } +} + +async function readOnlyContract(root: string, personaId: string): Promise { + const dir = path.join(root, personaId, 'compacted'); + const files = (await readdir(dir)).filter((f) => f.endsWith('.json')); + assert.equal(files.length, 1, `expected exactly one contract file, got ${files.join(', ')}`); + return JSON.parse(await readFile(path.join(dir, files[0]), 'utf8')) as CompactedTrajectoryContract; +} + +test('records a run and emits a contract-shaped artifact', async () => { + await withRoot(async (root) => { + const recorder = createTrajectoryRecorder({ + personaId: 'demo', + agentName: 'demo', + workspaceId: 'ws-1', + trajectoryRoot: root, + log: silentLog, + autoCompact: false + }); + + await recorder.begin(cronEvent); + await recorder.context.note('looked at the issue'); + await recorder.context.decide('lib A vs B', 'A', 'better types', [ + { option: 'B', reason: 'older' } + ]); + await recorder.context.done('did the thing', 0.9); + + const contract = await readOnlyContract(root, 'demo'); + + // Shape matches the locked A↔B contract. + assert.match(contract.id, /^traj_/); + assert.equal(contract.version, 1); + assert.equal(contract.personaId, 'demo'); + assert.equal(contract.projectId, 'ws-1'); // workspaceId fallback + assert.equal(contract.task.title, 'cron:weekly'); + assert.equal(contract.status, 'completed'); + assert.equal(typeof contract.startedAt, 'string'); + assert.equal(typeof contract.completedAt, 'string'); + + // Decisions flattened to {question,chosen,reasoning,alternatives[]}. + const decision = contract.decisions.find((d) => d.question === 'lib A vs B'); + assert.ok(decision, 'decision should be present'); + assert.equal(decision?.chosen, 'A'); + assert.equal(decision?.reasoning, 'better types'); + assert.deepEqual(decision?.alternatives, [{ option: 'B', reason: 'older' }]); + + // Retrospective projected to the contract subset. + assert.equal(contract.retrospective?.summary, 'did the thing'); + assert.equal(contract.retrospective?.confidence, 0.9); + assert.ok(Array.isArray(contract.retrospective?.learnings)); + + // NOT a `trail compact` aggregate — ai-hist's filter must keep it. + const raw = contract as unknown as Record; + assert.equal(raw.type, undefined); + assert.equal(raw.sourceTrajectories, undefined); + }); +}); + +test('auto-finalizes on complete() when the handler did not call done()', async () => { + await withRoot(async (root) => { + const recorder = createTrajectoryRecorder({ + personaId: 'demo', + agentName: 'demo', + workspaceId: 'ws-1', + trajectoryRoot: root, + log: silentLog, + autoCompact: false + }); + + await recorder.begin(cronEvent); + await recorder.context.note('worked without a self-assessed retrospective'); + await recorder.complete(); + + const contract = await readOnlyContract(root, 'demo'); + assert.equal(contract.status, 'completed'); + assert.ok(contract.retrospective, 'auto-finalize still emits a retrospective'); + assert.equal(contract.retrospective?.summary, 'Completed cron:weekly'); + }); +}); + +test('handler error abandons the run and emits an abandoned contract', async () => { + await withRoot(async (root) => { + const recorder = createTrajectoryRecorder({ + personaId: 'demo', + agentName: 'demo', + workspaceId: 'ws-1', + trajectoryRoot: root, + log: silentLog, + autoCompact: false + }); + + await recorder.begin(cronEvent); + await recorder.context.note('started'); + await recorder.fail(new Error('boom')); + + const contract = await readOnlyContract(root, 'demo'); + assert.equal(contract.status, 'abandoned'); + assert.equal(contract.completedAt !== null, true); + assert.equal(contract.retrospective, null); + }); +}); + +test('done() is idempotent — a later complete() does not double-emit', async () => { + await withRoot(async (root) => { + const recorder = createTrajectoryRecorder({ + personaId: 'demo', + agentName: 'demo', + workspaceId: 'ws-1', + trajectoryRoot: root, + log: silentLog, + autoCompact: false + }); + + await recorder.begin(cronEvent); + await recorder.context.done('handler self-finalized', 0.7); + await recorder.complete(); // should no-op + + const contract = await readOnlyContract(root, 'demo'); // asserts exactly one file + assert.equal(contract.retrospective?.summary, 'handler self-finalized'); + assert.equal(contract.retrospective?.confidence, 0.7); + }); +}); + +test('recordTrajectories:false disables recording (no files, safe no-ops)', async () => { + await withRoot(async (root) => { + const recorder = createTrajectoryRecorder({ + personaId: 'demo', + agentName: 'demo', + workspaceId: 'ws-1', + trajectoryRoot: root, + recordTrajectories: false, + log: silentLog + }); + + await recorder.begin(cronEvent); + await recorder.context.note('x'); + await recorder.context.decide('q', 'c', 'r'); + await recorder.context.done('y', 1); + + await assert.rejects(readdir(path.join(root, 'demo', 'compacted')), /ENOENT/); + }); +}); + +test('no resolvable trajectory root → recorder is a no-op', async () => { + await withRoot(async (root) => { + const recorder = createTrajectoryRecorder({ + personaId: 'demo', + agentName: 'demo', + workspaceId: 'ws-1', + env: {}, // no TRAJECTORY_ROOT, no explicit trajectoryRoot + log: silentLog + }); + + await recorder.begin(cronEvent); + await recorder.context.note('x'); + await recorder.context.done('y', 1); + + await assert.rejects(readdir(path.join(root, 'demo', 'compacted')), /ENOENT/); + }); +}); diff --git a/packages/runtime/src/trajectory.ts b/packages/runtime/src/trajectory.ts new file mode 100644 index 00000000..3f680add --- /dev/null +++ b/packages/runtime/src/trajectory.ts @@ -0,0 +1,371 @@ +import { mkdir, writeFile } from 'node:fs/promises'; +import { join } from 'node:path'; +import { + TrajectoryClient, + type Alternative, + type Decision, + type TrajectorySession, + type Trajectory +} from 'agent-trajectories/sdk'; +import type { + CompactedTrajectoryContract, + TrajectoryContext, + TrajectoryDecisionRecord, + WorkforceCtx, + WorkforceEvent +} from './types.js'; + +/** + * Per-run trajectory recorder. Wraps the `agent-trajectories` SDK so the + * runtime can auto-record a decision trajectory (the WHY) around every + * handler invocation and emit a compacted, contract-shaped artifact on + * completion. The compacted artifact is the A↔B interface consumed by the + * ai-hist trajectory sync source. + * + * Lifecycle (driven by the runner dispatch loop): + * begin(event) → [handler calls ctx.trajectory.note/decide/...] → complete() + * ↘ fail(err) + * + * `context` is the handler-facing surface attached to `ctx.trajectory`. + * Every recording operation is best-effort: a failure inside the recorder + * is logged and swallowed so it can never break the run it is observing. + */ +export interface TrajectoryRecorder { + /** Handler-facing surface attached to `ctx.trajectory`. */ + readonly context: TrajectoryContext; + /** Start a trajectory for a single run. */ + begin(event: WorkforceEvent): Promise; + /** Finish the active run (no-op if the handler already called `done`). */ + complete(): Promise; + /** Record a handler error and abandon the active run. */ + fail(error: unknown): Promise; +} + +export interface TrajectoryRecorderOptions { + /** Persona id — scopes the store and stamps every contract artifact. */ + personaId: string; + /** Logical agent name recorded as the trajectory's participant. */ + agentName: string; + /** Workspace the run belongs to (projectId fallback + tag). */ + workspaceId: string; + /** + * Persona `recordTrajectories` flag. `false` disables recording entirely; + * `undefined`/`true` keep it on (subject to a resolvable root). + */ + recordTrajectories?: boolean; + /** + * Explicit trajectory root. When omitted the recorder falls back to + * `env.TRAJECTORY_ROOT`. If neither resolves, recording is disabled — the + * runtime never silently writes to the process cwd. The cloud runtime sets + * this to the same value it passes to the ai-hist MCP so the MCP reads back + * exactly what the runtime wrote. + */ + trajectoryRoot?: string; + /** Explicit project id; otherwise derived from env then workspaceId. */ + projectId?: string; + /** Structured logger (warnings only — recording never throws). */ + log: WorkforceCtx['log']; + /** Environment source (defaults to `process.env`). */ + env?: NodeJS.ProcessEnv; + /** + * autoCompact passthrough for the SDK (runs `trail compact` on done). + * Defaults to `{ mechanical: true, markdown: true }`, or `false` when + * `WORKFORCE_TRAJECTORY_AUTOCOMPACT` is `0`/`false`. Tests pass `false`. + */ + autoCompact?: boolean | { mechanical?: boolean; markdown?: boolean }; +} + +const DEFAULT_CONFIDENCE = 0.8; +const AUTO_APPROACH = 'Auto-recorded by the workforce runtime.'; + +const NOOP_CONTEXT: TrajectoryContext = { + async chapter() {}, + async note() {}, + async decide() {}, + async error() {}, + async done() {} +}; + +const NOOP_RECORDER: TrajectoryRecorder = { + context: NOOP_CONTEXT, + async begin() {}, + async complete() {}, + async fail() {} +}; + +/** + * Build a trajectory recorder. Returns a no-op recorder when recording is + * disabled (`recordTrajectories === false`) or when no trajectory root can be + * resolved — both cases keep `ctx.trajectory.*` safe to call from any handler. + */ +export function createTrajectoryRecorder(options: TrajectoryRecorderOptions): TrajectoryRecorder { + const env = options.env ?? process.env; + if (options.recordTrajectories === false) return NOOP_RECORDER; + const root = (options.trajectoryRoot ?? env.TRAJECTORY_ROOT)?.trim(); + if (!root) return NOOP_RECORDER; + return new ActiveTrajectoryRecorder(options, env, root); +} + +class ActiveTrajectoryRecorder implements TrajectoryRecorder { + private readonly client: TrajectoryClient; + private readonly root: string; + private readonly personaId: string; + private readonly projectId: string | null; + private readonly log: WorkforceCtx['log']; + private initialized = false; + private session: TrajectorySession | null = null; + private finalized = false; + private currentEvent: WorkforceEvent | null = null; + readonly context: TrajectoryContext; + + constructor(options: TrajectoryRecorderOptions, env: NodeJS.ProcessEnv, root: string) { + this.root = root; + this.personaId = options.personaId; + this.log = options.log; + this.projectId = + firstNonEmpty( + env.TRAJECTORIES_PROJECT, + env.WORKFORCE_PROJECT_ID, + options.projectId, + options.workspaceId + ) ?? null; + this.client = new TrajectoryClient({ + dataDir: join(root, this.personaId), + defaultAgent: options.agentName, + ...(this.projectId ? { projectId: this.projectId } : {}), + autoSave: true, + autoCompact: options.autoCompact ?? resolveAutoCompact(env) + }); + this.context = { + chapter: (title) => this.safe(() => this.session?.chapter(title)), + note: (content) => this.safe(() => this.session?.note(content)), + decide: (question, chosen, reasoning, alternatives) => + this.safe(() => this.session?.decide(question, chosen, reasoning, alternatives)), + error: (content) => this.safe(() => this.session?.error(content)), + done: (summary, confidence) => this.finalizeCompleted(summary, confidence) + }; + } + + async begin(event: WorkforceEvent): Promise { + this.currentEvent = event; + this.session = null; + this.finalized = false; + try { + if (!this.initialized) { + await this.client.init(); + this.initialized = true; + } + // Clear a stale active trajectory left by a crashed prior run so + // `start()` does not reject with ACTIVE_TRAJECTORY_EXISTS. + const active = await this.client.getActive(); + if (active) { + const stale = await this.client.open(active.id); + await stale?.abandon('superseded by a new run'); + } + const { title, description } = describeEvent(event); + this.session = await this.client.start(title, { + ...(description ? { description } : {}), + workflowId: workflowIdFor(event), + tags: [`persona:${this.personaId}`, `workspace:${event.workspaceId}`, `source:${event.source}`] + }); + await this.session.chapter(`handle ${eventLabel(event)}`); + } catch (err) { + this.session = null; + this.warn('begin', err); + } + } + + async complete(): Promise { + if (!this.session || this.finalized) return; + const title = this.currentEvent ? describeEvent(this.currentEvent).title : 'run'; + await this.finalizeCompleted(`Completed ${title}`, DEFAULT_CONFIDENCE); + } + + async fail(error: unknown): Promise { + if (!this.session || this.finalized) return; + this.finalized = true; + const session = this.session; + this.session = null; + const message = error instanceof Error ? error.message : String(error); + try { + await session.error(message); + const trajectory = await session.abandon(message); + await this.emitContract(trajectory); + } catch (err) { + this.warn('fail', err); + } + } + + private async finalizeCompleted(summary: string, confidence: number): Promise { + if (!this.session || this.finalized) return; + this.finalized = true; + const session = this.session; + this.session = null; + try { + const trajectory = await session.complete({ + summary: summary.trim() || 'Run completed.', + approach: AUTO_APPROACH, + confidence: clampConfidence(confidence) + }); + await this.emitContract(trajectory); + } catch (err) { + this.warn('complete', err); + } + } + + private async emitContract(trajectory: Trajectory): Promise { + const contract = toContract(trajectory, this.personaId, this.projectId); + const dir = join(this.root, this.personaId, 'compacted'); + await mkdir(dir, { recursive: true }); + await writeFile(join(dir, `${trajectory.id}.json`), `${JSON.stringify(contract, null, 2)}\n`); + } + + private async safe(op: () => Promise | undefined): Promise { + if (!this.session || this.finalized) return; + try { + await op(); + } catch (err) { + this.warn('record', err); + } + } + + private warn(stage: string, err: unknown): void { + this.log('warn', `trajectory.${stage}.failed`, { + persona: this.personaId, + error: err instanceof Error ? err.message : String(err) + }); + } +} + +/** + * Project a completed/abandoned raw trajectory into the locked A↔B contract + * shape. Decisions are flattened from chapter `decision` events plus the + * retrospective; the retrospective is projected to its contract subset. + */ +export function toContract( + trajectory: Trajectory, + personaId: string, + projectId: string | null +): CompactedTrajectoryContract { + const retrospective = trajectory.retrospective + ? { + summary: trajectory.retrospective.summary, + approach: trajectory.retrospective.approach, + learnings: trajectory.retrospective.learnings ?? [], + confidence: trajectory.retrospective.confidence + } + : null; + + return { + id: trajectory.id, + version: trajectory.version, + personaId, + projectId: projectId ?? trajectory.projectId ?? null, + task: { + title: trajectory.task.title, + description: trajectory.task.description ?? null + }, + status: trajectory.status, + startedAt: trajectory.startedAt, + completedAt: trajectory.completedAt ?? null, + decisions: collectDecisions(trajectory), + retrospective + }; +} + +function collectDecisions(trajectory: Trajectory): TrajectoryDecisionRecord[] { + const out: TrajectoryDecisionRecord[] = []; + const seen = new Set(); + const push = (raw: unknown): void => { + const decision = normalizeDecision(raw); + if (!decision) return; + const key = `${decision.question}${decision.chosen}`; + if (seen.has(key)) return; + seen.add(key); + out.push(decision); + }; + for (const chapter of trajectory.chapters) { + for (const event of chapter.events) { + if (event.type === 'decision' && event.raw) push(event.raw); + } + } + for (const decision of trajectory.retrospective?.decisions ?? []) push(decision); + return out; +} + +function normalizeDecision(raw: unknown): TrajectoryDecisionRecord | null { + if (typeof raw !== 'object' || raw === null) return null; + const decision = raw as Partial; + return { + question: decision.question ?? '', + chosen: decision.chosen ?? '', + reasoning: decision.reasoning ?? '', + alternatives: normalizeAlternatives(decision.alternatives) + }; +} + +function normalizeAlternatives(alternatives: Decision['alternatives'] | undefined): Array<{ option: string; reason?: string }> { + if (!Array.isArray(alternatives)) return []; + return alternatives.map((alternative) => { + // The SDK schema accepts both string[] (legacy) and Alternative[]. + if (typeof alternative === 'string') return { option: alternative }; + const alt = alternative as Alternative; + return { + option: alt.option ?? '', + ...(alt.reason ? { reason: alt.reason } : {}) + }; + }); +} + +function describeEvent(event: WorkforceEvent): { title: string; description?: string } { + if (event.source === 'cron') { + return { title: `cron:${event.name}`, description: `cron schedule "${event.name}" (${event.cron})` }; + } + const title = event.summary?.title?.trim() || `${event.source}:${event.type}`; + return { title, description: `${event.source} ${event.type} — event ${event.id}` }; +} + +function eventLabel(event: WorkforceEvent): string { + return event.source === 'cron' ? 'cron:tick' : `${event.source}:${event.type}`; +} + +function workflowIdFor(event: WorkforceEvent): string { + return `run_${event.id.replace(/[^A-Za-z0-9_-]/g, '_')}_${event.attempt}`; +} + +function clampConfidence(value: number): number { + if (!Number.isFinite(value)) return 0.5; + return Math.min(1, Math.max(0, value)); +} + +function resolveAutoCompact(env: NodeJS.ProcessEnv): false | { mechanical: boolean; markdown: boolean } { + const disabled = env.WORKFORCE_TRAJECTORY_AUTOCOMPACT?.trim().toLowerCase(); + if (disabled === '0' || disabled === 'false') return false; + return { mechanical: true, markdown: true }; +} + +function firstNonEmpty(...values: Array): string | undefined { + for (const value of values) { + const trimmed = value?.trim(); + if (trimmed) return trimmed; + } + return undefined; +} + +const RECORDER_KEY = Symbol.for('workforce.runtime.trajectoryRecorder'); + +/** Attach a recorder to a ctx so the runner can drive its lifecycle. */ +export function attachTrajectoryRecorder(ctx: WorkforceCtx, recorder: TrajectoryRecorder): void { + Object.defineProperty(ctx, RECORDER_KEY, { + value: recorder, + enumerable: false, + configurable: true, + writable: false + }); +} + +/** Read the recorder the ctx was built with (no-op recorder when absent). */ +export function getTrajectoryRecorder(ctx: WorkforceCtx): TrajectoryRecorder { + const recorder = (ctx as unknown as Record)[RECORDER_KEY]; + return (recorder as TrajectoryRecorder | undefined) ?? NOOP_RECORDER; +} diff --git a/packages/runtime/src/types.ts b/packages/runtime/src/types.ts index 07445f18..6d4ac37d 100644 --- a/packages/runtime/src/types.ts +++ b/packages/runtime/src/types.ts @@ -321,6 +321,72 @@ export interface WorkforceDeploymentContext { readonly parentDeploymentId: string | null; } +/** A single decision in the compacted trajectory contract. */ +export interface TrajectoryDecisionRecord { + question: string; + chosen: string; + reasoning: string; + alternatives: Array<{ option: string; reason?: string }>; +} + +/** Retrospective subset carried in the compacted trajectory contract. */ +export interface TrajectoryRetrospectiveRecord { + summary: string; + approach: string; + learnings: string[]; + confidence: number; +} + +/** + * Compacted, contract-shaped trajectory artifact emitted once per completed + * (or abandoned) run. This is the frozen A↔B interface consumed by the + * ai-hist trajectory sync source — one JSON file per run at + * `$TRAJECTORY_ROOT//compacted/.json`. It deliberately omits + * the `type:"compacted"` / `sourceTrajectories[]` markers of `trail compact` + * aggregates so ai-hist's defensive ingest filter keeps it. + */ +export interface CompactedTrajectoryContract { + id: string; + version: number; + personaId: string; + projectId: string | null; + task: { title: string; description: string | null }; + status: 'active' | 'completed' | 'abandoned' | string; + startedAt: string; + completedAt: string | null; + decisions: TrajectoryDecisionRecord[]; + retrospective: TrajectoryRetrospectiveRecord | null; +} + +/** + * Auto-recording trajectory surface. Handlers narrate their decision + * trajectory (the WHY) through these methods; the runtime opens a trajectory + * around each run and emits the compacted contract on completion. Every method + * is safe to call even when recording is disabled (`recordTrajectories: false` + * or no resolvable `TRAJECTORY_ROOT`) — it then no-ops. + */ +export interface TrajectoryContext { + /** Open a new logical phase of work within the run. */ + chapter(title: string): Promise; + /** Record a free-form observation. */ + note(content: string): Promise; + /** Record a structured decision and the alternatives considered. */ + decide( + question: string, + chosen: string, + reasoning: string, + alternatives?: Array<{ option: string; reason?: string }> + ): Promise; + /** Record an error encountered while working. */ + error(content: string): Promise; + /** + * Finish the run with a retrospective. Optional — when the handler does not + * call it, the runner auto-finalizes on return. Idempotent: the first + * `done`/auto-finalize wins. + */ + done(summary: string, confidence: number): Promise; +} + /** * The context object handlers receive on every event invocation. * Provider data is accessed via the VFS helpers exported from the runtime @@ -356,6 +422,12 @@ export interface WorkforceCtx { workflow: WorkflowContext; /** Schedule one-off follow-up ticks. */ schedule: ScheduleContext; + /** + * Auto-recorded decision trajectory (the WHY). No-op when recording is + * disabled (`persona.recordTrajectories: false` or no resolvable + * `TRAJECTORY_ROOT`), so it is always safe to call from a handler. + */ + trajectory: TrajectoryContext; /** Structured logger; every line is forwarded to the gateway. */ log: (level: 'debug' | 'info' | 'warn' | 'error', message: string, attrs?: Record) => void; } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 2cf0c8e6..fc6dfd8f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -123,6 +123,9 @@ importers: '@relayfile/adapter-core': specifier: ^0.3.29 version: 0.3.29(@relayfile/sdk@0.7.40) + agent-trajectories: + specifier: ^0.5.3 + version: 0.5.3 packages/workload-router: dependencies: