apunkt/rowboat
1
0
Fork 0
mirror of https://github.com/rowboatlabs/rowboat.git synced 2026-07-12 21:02:17 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(x): stage 6 — migrate headless callers to the turn runtime

Browse source 
New src/agents/headless.ts wraps the turn runtime in the old headless
calling convention: startHeadlessAgent returns the turn id immediately
(callers record it in pointer files / bus events before completion) and
a done promise settling with { outcome, state, summary };
runHeadlessAgent awaits it. throwOnError reproduces the old
waitForRunCompletion({ throwOnError }) semantics via HeadlessRunError;
summary reproduces extractAgentResponse (last assistant text);
toolInputPaths replaces the run-bus tool-invocation subscriptions by
reading invoked calls from durable turn state. Model overrides pair the
caller's model id with the app-default provider. Unit-tested against an
injected fake runtime (7 tests).

Migrated all nine callers:
- background-tasks/runner: handle start wrapped in withUseCase so tools
  (notify-user) read the use case via AsyncLocalStorage
- knowledge/live-note/runner: same shape, gains withUseCase
- pre_built/runner, knowledge/agent_notes: run-and-wait
- knowledge/tag_notes, label_emails, build_graph: edited/created paths
  now come from turn state (toolInputPaths) instead of bus streaming
- knowledge/inline_tasks (both sites): summary text feeds the existing
  marker parsing unchanged
- agent-schedule/runner: fire-and-forget start with
  AbortSignal.timeout(TIMEOUT_MS); dropped the now-unused
  runsRepo/agentRuntime/idGenerator plumbing

Code-mode sessions remain on the runs infrastructure (deliberate
carve-out until stage 7 scoping); agents/utils.ts stays for
launch-code-task's extractAgentResponse. The notify-user useCase gate
already prefers ALS with a best-effort fetchRun fallback, so it works
unchanged for turn ids.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Ramnique Singh 2026-07-02 13:37:39 +05:30
parent 0042dde3bb
commit c73531ec38
11 changed files with 538 additions and 249 deletions
Download patch file Download diff file Expand all files Collapse all files

58
apps/x/packages/core/src/agent-schedule/runner.ts
View file

@ -2,13 +2,10 @@ import { CronExpressionParser } from "cron-parser";
import container from "../di/container.js";
import { IAgentScheduleRepo } from "./repo.js";
import { IAgentScheduleStateRepo } from "./state-repo.js";
import { IRunsRepo } from "../runs/repo.js";
import { IAgentRuntime } from "../agents/runtime.js";
import { IMonotonicallyIncreasingIdGenerator } from "../application/lib/id-gen.js";
import { AgentScheduleConfig, AgentScheduleEntry } from "@x/shared/dist/agent-schedule.js";
import { AgentScheduleState, AgentScheduleStateEntry } from "@x/shared/dist/agent-schedule-state.js";
import { MessageEvent } from "@x/shared/dist/runs.js";
import { createRun } from "../runs/runs.js";
import { startHeadlessAgent } from "../agents/headless.js";
import { withUseCase } from "../analytics/use_case.js";
import z from "zod";
const DEFAULT_STARTING_MESSAGE = "go";
@ -147,10 +144,7 @@ function shouldRunNow(
async function runAgent(
agentName: string,
entry: z.infer<typeof AgentScheduleEntry>,
stateRepo: IAgentScheduleStateRepo,
runsRepo: IRunsRepo,
agentRuntime: IAgentRuntime,
idGenerator: IMonotonicallyIncreasingIdGenerator
stateRepo: IAgentScheduleStateRepo
): Promise<void> {
console.log(`[AgentRunner] Starting agent: ${agentName}`);
@ -163,31 +157,24 @@ async function runAgent(
});
try {
// Create a new run via core (resolves agent + default model+provider).
const run = await createRun({
agentId: agentName,
useCase: 'copilot_chat',
subUseCase: 'scheduled',
});
console.log(`[AgentRunner] Created run ${run.id} for agent ${agentName}`);
// Add the starting message as a user message
// Start a standalone turn (resolves agent + default model/provider).
// Fire-and-forget like the old trigger(): the schedule state machine
// below runs on wall-clock; completion is observed only for logging.
// The signal caps a hung turn at the same TIMEOUT_MS the state-based
// watchdog uses.
const startingMessage = entry.startingMessage ?? DEFAULT_STARTING_MESSAGE;
const messageEvent: z.infer<typeof MessageEvent> = {
runId: run.id,
type: "message",
messageId: await idGenerator.next(),
message: {
role: "user",
content: startingMessage,
},
subflow: [],
};
await runsRepo.appendEvents(run.id, [messageEvent]);
console.log(`[AgentRunner] Sent starting message to agent ${agentName}: "${startingMessage}"`);
// Trigger the run
await agentRuntime.trigger(run.id);
const handle = await withUseCase(
{ useCase: 'copilot_chat', subUseCase: 'scheduled' },
() => startHeadlessAgent({
agentId: agentName,
message: startingMessage,
signal: AbortSignal.timeout(TIMEOUT_MS),
}),
);
console.log(`[AgentRunner] Started turn ${handle.turnId} for agent ${agentName}: "${startingMessage}"`);
void handle.done
.then((result) => console.log(`[AgentRunner] Turn ${handle.turnId} (${agentName}) settled: ${result.outcome.status}`))
.catch((error) => console.error(`[AgentRunner] Turn ${handle.turnId} (${agentName}) errored:`, error instanceof Error ? error.message : error));
// Calculate next run time
const nextRunAt = calculateNextRunAt(entry.schedule);
@ -264,9 +251,6 @@ async function checkForTimeouts(
async function pollAndRun(): Promise<void> {
const scheduleRepo = container.resolve<IAgentScheduleRepo>("agentScheduleRepo");
const stateRepo = container.resolve<IAgentScheduleStateRepo>("agentScheduleStateRepo");
const runsRepo = container.resolve<IRunsRepo>("runsRepo");
const agentRuntime = container.resolve<IAgentRuntime>("agentRuntime");
const idGenerator = container.resolve<IMonotonicallyIncreasingIdGenerator>("idGenerator");
// Load config and state
let config: z.infer<typeof AgentScheduleConfig>;
@ -314,7 +298,7 @@ async function pollAndRun(): Promise<void> {
if (shouldRunNow(entry, agentState)) {
// Run agent (don't await - let it run in background)
runAgent(agentName, entry, stateRepo, runsRepo, agentRuntime, idGenerator).catch((error) => {
runAgent(agentName, entry, stateRepo).catch((error) => {
console.error(`[AgentRunner] Unhandled error in runAgent for ${agentName}:`, error);
});
}

260
apps/x/packages/core/src/agents/headless.test.ts Normal file
View file

@ -0,0 +1,260 @@
import { describe, expect, it } from "vitest";
import type { z } from "zod";
import { reduceTurn, type TurnEvent } from "@x/shared/dist/turns.js";
import type {
CreateTurnInput,
ITurnRuntime,
TurnExecution,
TurnOutcome,
} from "../turns/api.js";
import {
HeadlessRunError,
assistantText,
lastAssistantText,
startHeadlessAgent,
toolInputPaths,
} from "./headless.js";
type TEvent = z.infer<typeof TurnEvent>;
const TURN_ID = "2026-07-02T10-00-00Z-0000001-000";
const TS = "2026-07-02T10:00:00Z";
const echoTool = {
toolId: "builtin:file-editText",
name: "file-editText",
description: "Edit",
inputSchema: {},
execution: "sync" as const,
requiresHuman: false,
};
function turnLog(opts: {
responseText?: string;
toolCalls?: Array<{ id: string; name: string; input: unknown; invoked: boolean }>;
failed?: string;
}): TEvent[] {
const events: TEvent[] = [
{
type: "turn_created",
schemaVersion: 1,
turnId: TURN_ID,
ts: TS,
sessionId: null,
agent: {
requested: { agentId: "worker" },
resolved: {
agentId: "worker",
systemPrompt: "SYS",
model: { provider: "fake", model: "m" },
tools: [echoTool],
},
},
context: [],
input: { role: "user", content: "go" },
config: { autoPermission: true, humanAvailable: false, maxModelCalls: 20 },
},
{
type: "model_call_requested",
turnId: TURN_ID,
ts: TS,
modelCallIndex: 0,
request: { messages: ["input"], parameters: {} },
},
];
const toolCalls = opts.toolCalls ?? [];
if (toolCalls.length > 0) {
events.push({
type: "model_call_completed",
turnId: TURN_ID,
ts: TS,
modelCallIndex: 0,
message: {
role: "assistant",
content: toolCalls.map((tc) => ({
type: "tool-call" as const,
toolCallId: tc.id,
toolName: tc.name,
arguments: tc.input as never,
})),
},
finishReason: "tool-calls",
usage: {},
});
for (const tc of toolCalls) {
if (!tc.invoked) continue;
events.push({
type: "tool_invocation_requested",
turnId: TURN_ID,
ts: TS,
toolCallId: tc.id,
toolId: "builtin:" + tc.name,
toolName: tc.name,
execution: "sync",
input: tc.input as never,
});
events.push({
type: "tool_result",
turnId: TURN_ID,
ts: TS,
toolCallId: tc.id,
toolName: tc.name,
source: "sync",
result: { output: "ok", isError: false },
});
}
}
if (opts.failed) {
events.push({
type: "model_call_failed",
turnId: TURN_ID,
ts: TS,
modelCallIndex: 0,
error: opts.failed,
});
events.push({ type: "turn_failed", turnId: TURN_ID, ts: TS, error: opts.failed, usage: {} });
} else if (opts.responseText !== undefined && toolCalls.length === 0) {
events.push({
type: "model_call_completed",
turnId: TURN_ID,
ts: TS,
modelCallIndex: 0,
message: { role: "assistant", content: opts.responseText },
finishReason: "stop",
usage: {},
});
events.push({
type: "turn_completed",
turnId: TURN_ID,
ts: TS,
output: { role: "assistant", content: opts.responseText },
finishReason: "stop",
usage: {},
});
}
return events;
}
class FakeRuntime implements ITurnRuntime {
createInputs: CreateTurnInput[] = [];
constructor(
private readonly log: TEvent[],
private readonly outcome: TurnOutcome,
) {}
async createTurn(input: CreateTurnInput): Promise<string> {
this.createInputs.push(input);
return TURN_ID;
}
advanceTurn(): TurnExecution {
return {
events: (async function* () {})(),
outcome: Promise.resolve(this.outcome),
};
}
async getTurn() {
return { turnId: TURN_ID, events: this.log };
}
}
const completedOutcome: TurnOutcome = {
status: "completed",
output: { role: "assistant", content: "done" },
finishReason: "stop",
usage: {},
};
describe("headless agent helpers", () => {
it("assistantText handles string and parts content", () => {
expect(assistantText({ role: "assistant", content: "hi" })).toBe("hi");
expect(
assistantText({
role: "assistant",
content: [
{ type: "reasoning", text: "hmm" },
{ type: "text", text: "a" },
{ type: "text", text: "b" },
],
}),
).toBe("ab");
expect(assistantText({ role: "assistant", content: "" })).toBeNull();
});
it("lastAssistantText returns the final assistant text in the transcript", () => {
const state = reduceTurn(turnLog({ responseText: "the summary" }));
expect(lastAssistantText(state)).toBe("the summary");
expect(lastAssistantText(reduceTurn(turnLog({ failed: "boom" })))).toBeNull();
});
it("toolInputPaths collects paths of invoked calls only", () => {
const state = reduceTurn(
turnLog({
toolCalls: [
{ id: "a", name: "file-editText", input: { path: "notes/x.md" }, invoked: true },
{ id: "b", name: "file-editText", input: { path: "notes/y.md" }, invoked: true },
{ id: "c", name: "file-writeText", input: { path: "notes/z.md" }, invoked: true },
{ id: "d", name: "file-editText", input: { path: "never-ran.md" }, invoked: false },
],
}),
);
expect(toolInputPaths(state, ["file-editText"])).toEqual(
new Set(["notes/x.md", "notes/y.md"]),
);
expect(toolInputPaths(state, ["file-writeText"])).toEqual(new Set(["notes/z.md"]));
});
});
describe("startHeadlessAgent", () => {
it("creates a standalone auto-permission turn and returns the id before settling", async () => {
const runtime = new FakeRuntime(turnLog({ responseText: "the summary" }), completedOutcome);
const handle = await startHeadlessAgent(
{ agentId: "worker", message: "go", model: "m", provider: "fake" },
runtime,
);
expect(handle.turnId).toBe(TURN_ID);
expect(runtime.createInputs[0]).toMatchObject({
agent: { agentId: "worker", overrides: { model: { provider: "fake", model: "m" } } },
sessionId: null,
context: [],
input: { role: "user", content: "go" },
config: { autoPermission: true, humanAvailable: false },
});
const result = await handle.done;
expect(result.outcome.status).toBe("completed");
expect(result.summary).toBe("the summary");
});
it("omits the model override when neither model nor provider is set", async () => {
const runtime = new FakeRuntime(turnLog({ responseText: "ok" }), completedOutcome);
await startHeadlessAgent({ agentId: "worker", message: "go" }, runtime);
expect(runtime.createInputs[0].agent.overrides).toBeUndefined();
});
it("throwOnError rejects done with HeadlessRunError on failed outcomes", async () => {
const runtime = new FakeRuntime(turnLog({ failed: "provider exploded" }), {
status: "failed",
error: "provider exploded",
usage: {},
});
const handle = await startHeadlessAgent(
{ agentId: "worker", message: "go", throwOnError: true },
runtime,
);
await expect(handle.done).rejects.toThrowError(HeadlessRunError);
await expect(handle.done).rejects.toThrowError("provider exploded");
});
it("without throwOnError a failed outcome resolves (old wait semantics)", async () => {
const runtime = new FakeRuntime(turnLog({ failed: "boom" }), {
status: "failed",
error: "boom",
usage: {},
});
const handle = await startHeadlessAgent({ agentId: "worker", message: "go" }, runtime);
const result = await handle.done;
expect(result.outcome.status).toBe("failed");
expect(result.summary).toBeNull();
});
});

159
apps/x/packages/core/src/agents/headless.ts Normal file
View file

@ -0,0 +1,159 @@
import type { z } from "zod";
import type { AssistantMessage } from "@x/shared/dist/message.js";
import {
reduceTurn,
type TurnState,
} from "@x/shared/dist/turns.js";
import container from "../di/container.js";
import { getDefaultModelAndProvider } from "../models/defaults.js";
import type { ITurnRuntime, TurnOutcome } from "../turns/api.js";
// Drop-in replacement for the old headless runs pattern
// (createRun → createMessage → waitForRunCompletion → extractAgentResponse):
// one standalone turn per invocation (sessionId null, automatic permissions,
// no human). startHeadlessAgent returns the turn id immediately (callers
// record it in pointer files / bus events before completion); `done` settles
// with the outcome, the reduced turn state, and the final assistant text.
export class HeadlessRunError extends Error {
constructor(
message: string,
readonly turnId: string,
readonly outcome: TurnOutcome,
) {
super(message);
this.name = "HeadlessRunError";
}
}
export interface HeadlessAgentOptions {
agentId: string;
message: string;
// Model id; when set without provider, the app-default provider applies.
model?: string;
provider?: string;
maxModelCalls?: number;
signal?: AbortSignal;
// Old waitForRunCompletion({ throwOnError: true }) semantics: `done`
// rejects with HeadlessRunError unless the turn completes.
throwOnError?: boolean;
}
export interface HeadlessAgentResult {
outcome: TurnOutcome;
state: TurnState;
// Last assistant text in the transcript (old extractAgentResponse).
summary: string | null;
}
export interface HeadlessAgentHandle {
turnId: string;
done: Promise<HeadlessAgentResult>;
}
export function assistantText(
message: z.infer<typeof AssistantMessage>,
): string | null {
const content = message.content;
if (typeof content === "string") {
return content || null;
}
const text = content
.map((part) => (part.type === "text" ? part.text : ""))
.join("");
return text || null;
}
export function lastAssistantText(state: TurnState): string | null {
for (let i = state.modelCalls.length - 1; i >= 0; i--) {
const response = state.modelCalls[i].response;
if (response) {
const text = assistantText(response);
if (text) {
return text;
}
}
}
return null;
}
// Paths passed to the given file tools across the turn — replaces the old
// pattern of subscribing to the run bus for tool-invocation events. Only
// actually-invoked calls count (denied/unknown calls never ran).
export function toolInputPaths(
state: TurnState,
toolNames: string[],
): Set<string> {
const paths = new Set<string>();
for (const toolCall of state.toolCalls) {
if (!toolNames.includes(toolCall.toolName) || !toolCall.invocation) {
continue;
}
const input = toolCall.input as { path?: unknown } | null | undefined;
if (input && typeof input === "object" && typeof input.path === "string") {
paths.add(input.path);
}
}
return paths;
}
export async function startHeadlessAgent(
options: HeadlessAgentOptions,
// Injectable for tests; defaults to the app container's runtime.
turnRuntime: ITurnRuntime = container.resolve<ITurnRuntime>("turnRuntime"),
): Promise<HeadlessAgentHandle> {
let modelOverride: { provider: string; model: string } | undefined;
if (options.model || options.provider) {
const defaults = await getDefaultModelAndProvider();
modelOverride = {
provider: options.provider ?? defaults.provider,
model: options.model ?? defaults.model,
};
}
const turnId = await turnRuntime.createTurn({
agent: {
agentId: options.agentId,
...(modelOverride ? { overrides: { model: modelOverride } } : {}),
},
sessionId: null,
context: [],
input: { role: "user", content: options.message },
config: {
autoPermission: true,
humanAvailable: false,
...(options.maxModelCalls === undefined
? {}
: { maxModelCalls: options.maxModelCalls }),
},
});
const execution = turnRuntime.advanceTurn(turnId, undefined, {
signal: options.signal,
});
const done = execution.outcome.then(async (outcome) => {
const state = reduceTurn((await turnRuntime.getTurn(turnId)).events);
if (options.throwOnError && outcome.status !== "completed") {
throw new HeadlessRunError(
outcome.status === "failed"
? outcome.error
: `turn ${outcome.status}`,
turnId,
outcome,
);
}
return { outcome, state, summary: lastAssistantText(state) };
});
// The handle may be used fire-and-forget; rejections surface when awaited.
done.catch(() => undefined);
return { turnId, done };
}
export async function runHeadlessAgent(
options: HeadlessAgentOptions,
turnRuntime?: ITurnRuntime,
): Promise<HeadlessAgentResult & { turnId: string }> {
const handle = await startHeadlessAgent(options, turnRuntime);
const result = await handle.done;
return { turnId: handle.turnId, ...result };
}

49
apps/x/packages/core/src/background-tasks/runner.ts
View file

@ -1,9 +1,8 @@
import type { BackgroundTask, BackgroundTaskTriggerType } from '@x/shared/dist/background-task.js';
import { PrefixLogger } from '@x/shared/dist/prefix-logger.js';
import { fetchTask, patchTask, prependRunId } from './fileops.js';
import { createRun, createMessage } from '../runs/runs.js';
import { getBackgroundTaskAgentModel } from '../models/defaults.js';
import { extractAgentResponse, waitForRunCompletion } from '../agents/utils.js';
import { startHeadlessAgent } from '../agents/headless.js';
import { buildTriggerBlock } from '../agents/build-trigger-block.js';
import { backgroundTaskBus } from './bus.js';
import { withUseCase } from '../analytics/use_case.js';
@ -139,20 +138,25 @@ export async function runBackgroundTask(
}
const model = task.model || await getBackgroundTaskAgentModel();
const agentRun = await createRun({
agentId: 'background-task-agent',
model,
...(task.provider ? { provider: task.provider } : {}),
useCase: 'background_task_agent',
// Granular trigger as analytics sub-use-case — matches live-note's
// pattern at runner.ts:149.
subUseCase: trigger,
});
// Establish the use-case context for the whole turn so every tool the
// agent calls (notably notify-user) reads `background_task_agent` via
// getCurrentUseCase(); the AsyncLocalStorage context set here flows
// through the turn's async execution chain.
const handle = await withUseCase(
{ useCase: 'background_task_agent', subUseCase: trigger },
() => startHeadlessAgent({
agentId: 'background-task-agent',
message: buildMessage(slug, task, trigger, context, codeProject),
model,
...(task.provider ? { provider: task.provider } : {}),
throwOnError: true,
}),
);
const runId = agentRun.id;
// Record this run in the task's runs.log pointer file (newest first).
// The transcript itself lives at the global $WorkDir/runs/<runId>.jsonl
// — runs.log is just an index that ties runIds to this task.
const runId = handle.turnId;
// Record this turn in the task's runs.log pointer file (newest first).
// The transcript itself lives at $WorkDir/storage/turns/YYYY/MM/DD/
// — runs.log is just an index that ties turn ids to this task.
await prependRunId(slug, runId);
const startedAt = new Date().toISOString();
@ -184,20 +188,7 @@ export async function runBackgroundTask(
});
try {
// Establish the use-case context for the whole run so every tool the
// agent calls (notably notify-user) reads `background_task_agent` via
// getCurrentUseCase(). createMessage synchronously fires
// agentRuntime.trigger(), so the detached run loop — and the tool
// calls within it — inherit this AsyncLocalStorage context. (The
// runtime's own enterUseCase runs inside an async generator and
// doesn't reliably propagate to tool execution, so we set it here at
// the trigger point instead.)
await withUseCase(
{ useCase: 'background_task_agent', subUseCase: trigger },
() => createMessage(runId, buildMessage(slug, task, trigger, context, codeProject)),
);
await waitForRunCompletion(runId, { throwOnError: true });
const summary = await extractAgentResponse(runId);
const { summary } = await handle.done;
// Success — bump cycle anchor, refresh summary, clear any prior error.
await patchTask(slug, {

12
apps/x/packages/core/src/knowledge/agent_notes.ts
View file

@ -2,9 +2,9 @@ import fs from 'fs';
import path from 'path';
import { google } from 'googleapis';
import { WorkDir } from '../config/config.js';
import { createRun, createMessage } from '../runs/runs.js';
import { runHeadlessAgent } from '../agents/headless.js';
import { getKgModel } from '../models/defaults.js';
import { getErrorDetails, waitForRunCompletion } from '../agents/utils.js';
import { getErrorDetails } from '../agents/utils.js';
import { serviceLogger } from '../services/service_logger.js';
import { loadUserConfig, updateUserEmail } from '../config/user_config.js';
import { GoogleClientFactory } from './google-client-factory.js';
@ -281,14 +281,12 @@ async function processAgentNotes(): Promise<void> {
const timestamp = new Date().toISOString();
const message = `Current timestamp: ${timestamp}\n\nProcess the following source material and update the Agent Notes folder accordingly.\n\n${messageParts.join('\n\n')}`;
const agentRun = await createRun({
await runHeadlessAgent({
agentId: AGENT_ID,
message,
model: await getKgModel(),
useCase: 'knowledge_sync',
subUseCase: 'agent_notes',
throwOnError: true,
});
await createMessage(agentRun.id, message);
await waitForRunCompletion(agentRun.id, { throwOnError: true });
// Mark everything as processed
for (const p of emailPaths) {

58
apps/x/packages/core/src/knowledge/build_graph.ts
View file

@ -2,9 +2,8 @@ import fs from 'fs';
import path from 'path';
import { WorkDir } from '../config/config.js';
import { getKgModel } from '../models/defaults.js';
import { createRun, createMessage } from '../runs/runs.js';
import { bus } from '../runs/bus.js';
import { getErrorDetails, waitForRunCompletion } from '../agents/utils.js';
import { runHeadlessAgent, toolInputPaths } from '../agents/headless.js';
import { getErrorDetails } from '../agents/utils.js';
import { serviceLogger, type ServiceRunContext } from '../services/service_logger.js';
import {
loadState,
@ -89,14 +88,6 @@ function hasNoiseLabels(content: string): boolean {
return false;
}
function extractPathFromToolInput(input: string): string | null {
try {
const parsed = JSON.parse(input) as { path?: string };
return typeof parsed.path === 'string' ? parsed.path : null;
} catch {
return null;
}
}
function ensureSuggestedTopicsFileLocation(): string {
if (fs.existsSync(SUGGESTED_TOPICS_PATH)) {
@ -252,13 +243,6 @@ async function createNotesFromBatch(
fs.mkdirSync(NOTES_OUTPUT_DIR, { recursive: true });
}
// Create a run for the note creation agent
const run = await createRun({
agentId: NOTE_CREATION_AGENT,
model: await getKgModel(),
useCase: 'knowledge_sync',
subUseCase: 'build_graph',
});
const suggestedTopicsContent = readSuggestedTopicsFile();
// Build message with index and all files in the batch
@ -293,37 +277,19 @@ async function createNotesFromBatch(
message += `\n\n---\n\n`;
});
const notesCreated = new Set<string>();
const notesModified = new Set<string>();
const unsubscribe = await bus.subscribe(run.id, async (event) => {
if (event.type !== "tool-invocation") {
return;
}
if (event.toolName !== "file-writeText" && event.toolName !== "file-editText") {
return;
}
const toolPath = extractPathFromToolInput(event.input);
if (!toolPath) {
return;
}
if (event.toolName === "file-writeText") {
notesCreated.add(toolPath);
} else if (event.toolName === "file-editText") {
notesModified.add(toolPath);
}
const { turnId, state } = await runHeadlessAgent({
agentId: NOTE_CREATION_AGENT,
message,
model: await getKgModel(),
throwOnError: true,
});
await createMessage(run.id, message);
// Created/modified paths come from the durable turn state instead of
// streaming bus subscriptions.
const notesCreated = toolInputPaths(state, ["file-writeText"]);
const notesModified = toolInputPaths(state, ["file-editText"]);
// Wait for the run to complete
try {
await waitForRunCompletion(run.id, { throwOnError: true });
} finally {
unsubscribe();
}
return { runId: run.id, notesCreated, notesModified };
return { runId: turnId, notesCreated, notesModified };
}
/**

35
apps/x/packages/core/src/knowledge/inline_tasks.ts
View file

@ -3,13 +3,12 @@ import path from 'path';
import { CronExpressionParser } from 'cron-parser';
import { generateText } from 'ai';
import { WorkDir } from '../config/config.js';
import { createRun, createMessage, fetchRun } from '../runs/runs.js';
import { runHeadlessAgent } from '../agents/headless.js';
import { getKgModel } from '../models/defaults.js';
import container from '../di/container.js';
import type { IModelConfigRepo } from '../models/repo.js';
import { createProvider } from '../models/models.js';
import { inlineTask } from '@x/shared';
import { extractAgentResponse, waitForRunCompletion } from '../agents/utils.js';
import { captureLlmUsage } from '../analytics/usage.js';
import { withUseCase } from '../analytics/use_case.js';
@ -470,13 +469,6 @@ async function processInlineTasks(): Promise<void> {
console.log(`[InlineTasks] Running task: "${task.instruction.slice(0, 80)}..."`);
try {
const run = await createRun({
agentId: INLINE_TASK_AGENT,
model: await getKgModel(),
useCase: 'knowledge_sync',
subUseCase: 'inline_task_run',
});
const message = [
`Execute the following instruction from the note "${relativePath}":`,
'',
@ -488,10 +480,11 @@ async function processInlineTasks(): Promise<void> {
'```',
].join('\n');
await createMessage(run.id, message);
await waitForRunCompletion(run.id);
const result = await extractAgentResponse(run.id);
const { summary: result } = await runHeadlessAgent({
agentId: INLINE_TASK_AGENT,
message,
model: await getKgModel(),
});
if (result) {
if (task.targetId) {
// Recurring task with target region — replace content inside the region
@ -555,13 +548,6 @@ export async function processRowboatInstruction(
scheduleLabel: string | null;
response: string | null;
}> {
const run = await createRun({
agentId: INLINE_TASK_AGENT,
model: await getKgModel(),
useCase: 'knowledge_sync',
subUseCase: 'inline_task_run',
});
const message = [
`Process the following @rowboat instruction from the note "${notePath}":`,
'',
@ -573,10 +559,11 @@ export async function processRowboatInstruction(
'```',
].join('\n');
await createMessage(run.id, message);
await waitForRunCompletion(run.id);
const rawResponse = await extractAgentResponse(run.id);
const { summary: rawResponse } = await runHeadlessAgent({
agentId: INLINE_TASK_AGENT,
message,
model: await getKgModel(),
});
if (!rawResponse) {
return { instruction, schedule: null, scheduleLabel: null, response: null };
}

45
apps/x/packages/core/src/knowledge/label_emails.ts
View file

@ -1,10 +1,9 @@
import fs from 'fs';
import path from 'path';
import { WorkDir } from '../config/config.js';
import { createRun, createMessage } from '../runs/runs.js';
import { runHeadlessAgent, toolInputPaths } from '../agents/headless.js';
import { getKgModel } from '../models/defaults.js';
import { bus } from '../runs/bus.js';
import { getErrorDetails, waitForRunCompletion } from '../agents/utils.js';
import { getErrorDetails } from '../agents/utils.js';
import { serviceLogger } from '../services/service_logger.js';
import { limitEventItems } from './limit_event_items.js';
import {
@ -70,13 +69,6 @@ function getUnlabeledEmails(state: LabelingState): string[] {
async function labelEmailBatch(
files: { path: string; content: string }[]
): Promise<{ runId: string; filesEdited: Set<string> }> {
const run = await createRun({
agentId: LABELING_AGENT,
model: await getKgModel(),
useCase: 'knowledge_sync',
subUseCase: 'label_emails',
});
let message = `Label the following ${files.length} email files by prepending YAML frontmatter.\n\n`;
message += `**Important:** Use workspace-relative paths with file-editText (e.g. "gmail_sync/email.md", NOT absolute paths).\n\n`;
@ -92,33 +84,16 @@ async function labelEmailBatch(
message += `\n\n---\n\n`;
}
const filesEdited = new Set<string>();
const unsubscribe = await bus.subscribe(run.id, async (event) => {
if (event.type !== 'tool-invocation') {
return;
}
if (event.toolName !== 'file-editText') {
return;
}
try {
const parsed = JSON.parse(event.input) as { path?: string };
if (typeof parsed.path === 'string') {
filesEdited.add(parsed.path);
}
} catch {
// ignore parse errors
}
const { turnId, state } = await runHeadlessAgent({
agentId: LABELING_AGENT,
message,
model: await getKgModel(),
throwOnError: true,
});
await createMessage(run.id, message);
try {
await waitForRunCompletion(run.id, { throwOnError: true });
} finally {
unsubscribe();
}
return { runId: run.id, filesEdited };
// Edited paths come from the durable turn state instead of streaming
// bus subscriptions.
return { runId: turnId, filesEdited: toolInputPaths(state, ['file-editText']) };
}
/**

42
apps/x/packages/core/src/knowledge/live-note/runner.ts
View file

@ -1,8 +1,8 @@
import type { LiveNote, LiveNoteTriggerType } from '@x/shared/dist/live-note.js';
import { fetchLiveNote, patchLiveNote, readNoteBody } from './fileops.js';
import { createRun, createMessage } from '../../runs/runs.js';
import { getLiveNoteAgentModel } from '../../models/defaults.js';
import { extractAgentResponse, waitForRunCompletion } from '../../agents/utils.js';
import { startHeadlessAgent } from '../../agents/headless.js';
import { withUseCase } from '../../analytics/use_case.js';
import { buildTriggerBlock } from '../../agents/build-trigger-block.js';
import { liveNoteBus } from './bus.js';
import { PrefixLogger } from '@x/shared/dist/prefix-logger.js';
@ -109,17 +109,20 @@ export async function runLiveNoteAgent(
const bodyBefore = await readNoteBody(filePath);
const model = live.model ?? await getLiveNoteAgentModel();
const agentRun = await createRun({
agentId: 'live-note-agent',
model,
...(live.provider ? { provider: live.provider } : {}),
useCase: 'live_note_agent',
// Use the granular trigger as the analytics sub-use-case so
// dashboards can break down agent runs by what woke them up
// (manual / cron / window / event). Pass 1 routing emits the
// separate `routing` sub-use-case from routing.ts.
subUseCase: trigger,
});
// The use-case context propagates to every tool the agent calls; the
// granular trigger doubles as the sub-use-case (manual / cron /
// window / event) so dashboards can break down what woke the agent.
const handle = await withUseCase(
{ useCase: 'live_note_agent', subUseCase: trigger },
() => startHeadlessAgent({
agentId: 'live-note-agent',
message: buildMessage(filePath, live, trigger, context),
model,
...(live.provider ? { provider: live.provider } : {}),
throwOnError: true,
}),
);
const agentRun = { id: handle.turnId };
log.log(`${filePath} — start trigger=${trigger} runId=${agentRun.id}`);
@ -141,14 +144,11 @@ export async function runLiveNoteAgent(
});
try {
await createMessage(agentRun.id, buildMessage(filePath, live, trigger, context));
// throwOnError: surface any error event in the run's log (LLM API
// failures, tool errors, billing/credit issues) as a rejection so
// the failure branch records lastRunError. Without this the run
// can "complete" with errors silently and we'd hit the success
// branch with an empty summary, clobbering any prior lastRunError.
await waitForRunCompletion(agentRun.id, { throwOnError: true });
const summary = await extractAgentResponse(agentRun.id);
// throwOnError: a failed/cancelled turn rejects here so the
// failure branch records lastRunError. Without this the turn
// could settle silently and we'd hit the success branch with an
// empty summary, clobbering any prior lastRunError.
const { summary } = await handle.done;
const bodyAfter = await readNoteBody(filePath);
const didUpdate = bodyAfter !== bodyBefore;

45
apps/x/packages/core/src/knowledge/tag_notes.ts
View file

@ -1,10 +1,9 @@
import fs from 'fs';
import path from 'path';
import { WorkDir } from '../config/config.js';
import { createRun, createMessage } from '../runs/runs.js';
import { runHeadlessAgent, toolInputPaths } from '../agents/headless.js';
import { getKgModel } from '../models/defaults.js';
import { bus } from '../runs/bus.js';
import { getErrorDetails, waitForRunCompletion } from '../agents/utils.js';
import { getErrorDetails } from '../agents/utils.js';
import { serviceLogger } from '../services/service_logger.js';
import { limitEventItems } from './limit_event_items.js';
import {
@ -83,13 +82,6 @@ function getUntaggedNotes(state: NoteTaggingState): string[] {
async function tagNoteBatch(
files: { path: string; content: string }[]
): Promise<{ runId: string; filesEdited: Set<string> }> {
const run = await createRun({
agentId: NOTE_TAGGING_AGENT,
model: await getKgModel(),
useCase: 'knowledge_sync',
subUseCase: 'tag_notes',
});
let message = `Tag the following ${files.length} knowledge notes by prepending YAML frontmatter with appropriate tags.\n\n`;
message += `**Important:** Use workspace-relative paths with file-editText (e.g. "knowledge/People/Sarah Chen.md", NOT absolute paths).\n\n`;
@ -105,33 +97,16 @@ async function tagNoteBatch(
message += `\n\n---\n\n`;
}
const filesEdited = new Set<string>();
const unsubscribe = await bus.subscribe(run.id, async (event) => {
if (event.type !== 'tool-invocation') {
return;
}
if (event.toolName !== 'file-editText') {
return;
}
try {
const parsed = JSON.parse(event.input) as { path?: string };
if (typeof parsed.path === 'string') {
filesEdited.add(parsed.path);
}
} catch {
// ignore parse errors
}
const { turnId, state } = await runHeadlessAgent({
agentId: NOTE_TAGGING_AGENT,
message,
model: await getKgModel(),
throwOnError: true,
});
await createMessage(run.id, message);
try {
await waitForRunCompletion(run.id, { throwOnError: true });
} finally {
unsubscribe();
}
return { runId: run.id, filesEdited };
// Edited paths come from the durable turn state instead of streaming
// bus subscriptions.
return { runId: turnId, filesEdited: toolInputPaths(state, ['file-editText']) };
}
/**

24
apps/x/packages/core/src/pre_built/runner.ts
View file

@ -1,9 +1,8 @@
import fs from 'fs';
import path from 'path';
import { WorkDir } from '../config/config.js';
import { createRun, createMessage } from '../runs/runs.js';
import { runHeadlessAgent } from '../agents/headless.js';
import { getKgModel } from '../models/defaults.js';
import { waitForRunCompletion } from '../agents/utils.js';
import {
loadConfig,
loadState,
@ -38,15 +37,6 @@ async function runAgent(agentName: string): Promise<void> {
}
try {
// Create a run for the agent
// The agent file is expected to be in the agents directory with the same name
const run = await createRun({
agentId: agentName,
model: await getKgModel(),
useCase: 'knowledge_sync',
subUseCase: 'pre_built',
});
// Build trigger message with user context
const message = `Run your scheduled task.
@ -59,10 +49,14 @@ async function runAgent(agentName: string): Promise<void> {
Process new items and use the user context above to identify yourself when drafting responses.`;
await createMessage(run.id, message);
// Wait for completion
await waitForRunCompletion(run.id);
// The agent file is expected to be in the agents directory with
// the same name. Waits for the turn to settle (errors tolerated,
// matching the old no-throwOnError wait).
await runHeadlessAgent({
agentId: agentName,
message,
model: await getKgModel(),
});
// Update last run time
setLastRunTime(agentName, new Date());