From 69e991ff93b71689da97e127b08c1d9b68ad2cd2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 8 Oct 2026 09:16:48 -0400 Subject: [PATCH] Q52: goal closures stop linked work, and running sessions hear goal changes Co-Authored-By: Claude Opus 5.5 --- src/main/services/goal-change-poller.ts | 33 ++++++++++ src/main/services/heartbeat-scheduler.ts | 1 + src/main/services/task-execution-service.ts | 12 +++- tests/integration/goal-changes.test.ts | 69 +++++++++++++++++++++ tests/unit/goal-change-poller.test.ts | 20 ++++++ 5 files changed, 134 insertions(+), 1 deletion(-) create mode 100644 src/main/services/goal-change-poller.ts create mode 100644 tests/integration/goal-changes.test.ts create mode 100644 tests/unit/goal-change-poller.test.ts diff --git a/src/main/services/goal-change-poller.ts b/src/main/services/goal-change-poller.ts new file mode 100644 index 00000000..b57a4be6 --- /dev/null +++ b/src/main/services/goal-change-poller.ts @@ -0,0 +1,33 @@ +import type { DatabaseSync } from "node:sqlite"; + +interface GoalState { + title: string; + description: string; + status: string; +} + +/** Returns a notice whenever the run's goal is rewritten or closed after the run started, once per change. */ +export function createGoalChangePoller(db: DatabaseSync, goalId: string | null): () => string | null { + if (!goalId) return () => null; + const read = () => db.prepare("select title, description, status from goals where id = ?").get(goalId) as GoalState | undefined; + let seen = read(); + return () => { + const now = read(); + if (!seen || !now || (now.title === seen.title && now.description === seen.description && now.status === seen.status)) return null; + const before = seen; + seen = now; + console.info(`[task-run] goal ${goalId} changed mid-run (${now.status}); notifying the session`); + if (now.status !== before.status && now.status !== "active") { + return `The goal for this work ("${now.title}") is now ${now.status}. Stop starting new work on it: wrap up, report what you finished, and set this task's status.`; + } + return `The goal for this work changed while you were working. It now reads: "${now.title}" — ${now.description} Re-check your plan against it.`; + }; +} + +/** Joins mid-run notices from several sources into one follow-up, or null when none has news. */ +export function combineMidRunPolls(...polls: Array<() => string | null>): () => string | null { + return () => { + const notices = polls.map((poll) => poll()).filter((notice): notice is string => notice !== null); + return notices.length > 0 ? notices.join("\n\n") : null; + }; +} diff --git a/src/main/services/heartbeat-scheduler.ts b/src/main/services/heartbeat-scheduler.ts index 7a049fd5..9b650975 100644 --- a/src/main/services/heartbeat-scheduler.ts +++ b/src/main/services/heartbeat-scheduler.ts @@ -53,6 +53,7 @@ export class HeartbeatScheduler { and coalesce(u.request_count, 0) < p.monthly_request_limit and not exists (select 1 from task_runs r where r.task_id = t.id and r.state in ('queued', 'running')) and not exists (select 1 from tasks c where c.parent_task_id = t.id and c.status <> 'done') + and (t.origin is not null or t.goal_id is null or exists (select 1 from goals g where g.id = t.goal_id and g.status = 'active')) and (a.heartbeat_interval_sec <= 0 or not exists ( select 1 from task_runs hb where hb.agent_id = a.id and hb.trigger = 'heartbeat' and hb.state <> 'superseded' and hb.created_at > strftime('%Y-%m-%dT%H:%M:%fZ', ?, '-' || a.heartbeat_interval_sec || ' seconds') diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 308749e6..e1306489 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -11,6 +11,7 @@ import { deliverChatReplyIfMissing } from "./chat-agent-wake-service"; import { SkillService } from "./skill-service"; import { runAgentLoop, runTools } from "./agent-run-loop"; import type { RunModel } from "./skill-runtime"; +import { combineMidRunPolls, createGoalChangePoller } from "./goal-change-poller"; import type { AgentToolContext } from "./agent-tools"; import { buildAgentInstructions, @@ -121,6 +122,10 @@ export class TaskExecutionService { if (!sameWorkspaceAccess(run.workspace_mount_json, workspaceAccess(task, workspace))) { return this.supersedeRun(orgId, runId, run, "the task's workspace access changed after this run was queued."); } + const goal = task.origin === null && task.goal_id + ? this.db.prepare("select status from goals where id = ?").get(task.goal_id) as { status: string } | undefined + : undefined; + if (goal && goal.status !== "active") return this.supersedeRun(orgId, runId, run, `the task's goal is ${goal.status}, so its work stops.`); const openSubtask = this.db.prepare("select title from tasks where parent_task_id = ? and status <> 'done' limit 1").get(task.id) as { title: string } | undefined; if (openSubtask) return this.supersedeRun(orgId, runId, run, `the task has open subtasks ("${openSubtask.title}"); it runs again once they are done.`); const approvalProblem = executionApprovalProblem(this.db, orgId, task); @@ -187,6 +192,8 @@ export class TaskExecutionService { .toolkitsForAgent(orgId, agent.id) .map(connectorDisplayName); const language = this.appLanguage(); + // Baseline taken before any await, so a goal closed while tools load still reaches the session. + const goalChanges = createGoalChangePoller(this.db, task.goal_id); const externalTools = await loadComposioTools(this.db, this.providers.integrationKeys, orgId, agent.id); const result = await runAgentLoop( { @@ -223,7 +230,10 @@ export class TaskExecutionService { externalTools, onModelTurn: contextRecorder.commit, inputBudgetTokens: inputBudgetTokens(provider.context_window_tokens), - pollNewMessages: createRunMessagePoller(this.db, { orgId, onInjected: contextRecorder.addMidRun, agentId: agent.id, sourceThreadId: chatContext?.threadId ?? null, runStartedAt }), + pollNewMessages: combineMidRunPolls( + goalChanges, + createRunMessagePoller(this.db, { orgId, onInjected: contextRecorder.addMidRun, agentId: agent.id, sourceThreadId: chatContext?.threadId ?? null, runStartedAt }), + ), }, ); providerRequestId = result.lastRequestId; diff --git a/tests/integration/goal-changes.test.ts b/tests/integration/goal-changes.test.ts new file mode 100644 index 00000000..9e32cd22 --- /dev/null +++ b/tests/integration/goal-changes.test.ts @@ -0,0 +1,69 @@ +import { randomUUID } from "node:crypto"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, test } from "vitest"; +import { DatabaseConnection } from "../../src/main/database/connection"; +import { OnboardingService } from "../../src/main/services/onboarding-service"; +import { ProviderService } from "../../src/main/services/provider-service"; +import { providerAdapters } from "../../src/main/services/provider-adapters"; +import { TaskExecutionService } from "../../src/main/services/task-execution-service"; +import { claimNextQueuedRun } from "../../src/main/app/task-queue-worker"; +import type { AgentTurnResult, ProviderToolTurnInput } from "../../src/main/services/provider-tool-turn"; + +const testVault = { + encryptSecret: (secret: string) => Buffer.from(secret, "utf8"), + decryptSecret: (ciphertext: Buffer) => ciphertext.toString("utf8"), +}; + +function company(turn: (input: ProviderToolTurnInput, db: DatabaseConnection["db"]) => AgentTurnResult) { + const connection = new DatabaseConnection(join(tmpdir(), `ac-goalchg-${randomUUID()}.db`)); + const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, async (input) => turn(input, connection.db)); + const onboarding = new OnboardingService({ db: connection.db, saveProviderCredential: (input) => providers.saveCredential(input) }); + const execution = new TaskExecutionService(connection.db, providers); + let orgId = ""; + connection.transaction(() => { + orgId = onboarding.bootstrap({ + companyName: "FlowPilot", companyDescription: "CSM analytics.", employeeName: "Yang", workspacePath: "", + executionMode: "ai_ceo", autoApproveHires: true, reviewDeliverables: false, + goalTitle: "Ship the MVP.", goalDescription: "A working product.", + providerKey: "openai", providerSecret: "sk-test-key", providerModel: "gpt-5.4", + }).orgId; + }); + const goalId = (connection.db.prepare("select id from goals where org_id = ?").get(orgId) as { id: string }).id; + return { connection, execution, orgId, goalId }; +} + +describe("goal changes", () => { + test("work under an achieved goal stops: its queued run is superseded and the heartbeat leaves it alone", async () => { + let calls = 0; + const h = company(() => { calls += 1; return { text: "done", toolCalls: [], rawResponseJson: "{}" }; }); + const kickoff = claimNextQueuedRun(h.connection)!; + h.connection.db.prepare("update goals set status = 'achieved' where id = ?").run(h.goalId); + await h.execution.executeRun(h.orgId, kickoff.id); + expect(h.connection.db.prepare("select state, error_message from task_runs where id = ?").get(kickoff.id)) + .toEqual({ state: "superseded", error_message: "Superseded: the task's goal is achieved, so its work stops." }); + expect(h.connection.transaction(() => h.execution.enqueueHeartbeatRuns())).toEqual([]); + expect(calls).toBe(0); + h.connection.close(); + }); + + test("a running session hears about a goal rewrite before its next step", async () => { + let followUp: string | undefined; + let goalId = ""; + const h = company((input, db) => { + const last = input.messages.at(-1); + if (last?.role === "tool") { + followUp = last.followUp; + return { text: "Re-planned.", toolCalls: [], rawResponseJson: "{}" }; + } + // Same updated_at on purpose: the notice must follow the content, not the timestamp. + db.prepare("update goals set title = 'Ship the paid beta.', description = 'Charge from day one.' where id = ?").run(goalId); + return { text: "Checking messages.", toolCalls: [{ id: "c1", name: "read_messages", arguments: {} }], rawResponseJson: "{}" }; + }); + goalId = h.goalId; + const kickoff = claimNextQueuedRun(h.connection)!; + await h.execution.executeRun(h.orgId, kickoff.id); + expect(followUp).toContain('The goal for this work changed while you were working. It now reads: "Ship the paid beta." — Charge from day one. Re-check your plan against it.'); + h.connection.close(); + }); +}); diff --git a/tests/unit/goal-change-poller.test.ts b/tests/unit/goal-change-poller.test.ts new file mode 100644 index 00000000..12293ce3 --- /dev/null +++ b/tests/unit/goal-change-poller.test.ts @@ -0,0 +1,20 @@ +import { DatabaseSync } from "node:sqlite"; +import { describe, expect, test } from "vitest"; +import { combineMidRunPolls, createGoalChangePoller } from "../../src/main/services/goal-change-poller"; + +describe("goal change poller", () => { + test("reports a closure and a rewrite once each, against the state seen when the run began", () => { + const db = new DatabaseSync(":memory:"); + db.exec("create table goals(id text primary key, title text, description text, status text); insert into goals values ('g1', 'Ship', 'MVP', 'active');"); + const poll = createGoalChangePoller(db, "g1"); + expect(poll()).toBeNull(); + db.exec("update goals set status = 'achieved' where id = 'g1'"); + expect(poll()).toBe('The goal for this work ("Ship") is now achieved. Stop starting new work on it: wrap up, report what you finished, and set this task\'s status.'); + expect(poll()).toBeNull(); + db.exec("update goals set description = 'Paid beta' where id = 'g1'"); + expect(poll()).toContain("It now reads: \"Ship\" — Paid beta"); + expect(createGoalChangePoller(db, null)()).toBeNull(); + expect(combineMidRunPolls(() => "a", () => null, () => "b")()).toBe("a\n\nb"); + db.close(); + }); +});