From 305296fe4107c21c04e8ea05c250025663949499 Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Fri, 21 Aug 2026 09:32:27 +0800 Subject: [PATCH 1/2] fix(desktop): reconcile goal arm across reconnects Generated-by: Codex --- .../main/__tests__/goal-arm-outcome.test.ts | 81 ++++ .../src/main/__tests__/goal-dialog.test.tsx | 251 +++++++++++ .../__tests__/goal-preload-bridge.test.ts | 70 +++ ...runtime-host-reconnecting-ipc-main.test.ts | 219 +++++++++- ...time-host-session-domains-ipc-main.test.ts | 412 +++++++++++++++++- apps/desktop/src/main/ipc-reconnect-policy.ts | 46 ++ .../main/runtime-host-desktop-candidate.ts | 41 +- .../runtime-host-reconnecting-ipc-main.ts | 93 +++- .../runtime-host-session-domains-ipc-main.ts | 101 ++++- apps/desktop/src/preload/bridge-contract.d.ts | 2 +- apps/desktop/src/preload/preload.ts | 49 +-- .../preload/projected-session-runtime-host.ts | 57 +++ apps/desktop/src/renderer/goal-arm-outcome.ts | 31 ++ apps/desktop/src/renderer/goal-dialog.tsx | 62 ++- .../src/renderer/locales/shell-copy.ts | 43 ++ apps/desktop/src/shared/goal-arm.ts | 11 + 16 files changed, 1500 insertions(+), 69 deletions(-) create mode 100644 apps/desktop/src/main/__tests__/goal-arm-outcome.test.ts create mode 100644 apps/desktop/src/main/__tests__/goal-dialog.test.tsx create mode 100644 apps/desktop/src/main/__tests__/goal-preload-bridge.test.ts create mode 100644 apps/desktop/src/preload/projected-session-runtime-host.ts create mode 100644 apps/desktop/src/renderer/goal-arm-outcome.ts diff --git a/apps/desktop/src/main/__tests__/goal-arm-outcome.test.ts b/apps/desktop/src/main/__tests__/goal-arm-outcome.test.ts new file mode 100644 index 0000000000..c327a6fd93 --- /dev/null +++ b/apps/desktop/src/main/__tests__/goal-arm-outcome.test.ts @@ -0,0 +1,81 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import type { GoalArmOutcome } from '../../shared/goal-arm.js'; +import { interpretGoalArmOutcome } from '../../renderer/goal-arm-outcome.js'; +import { getShellCopy } from '../../renderer/locales/shell-copy.js'; + +test('successful Goal arming closes while every reconciliation result locks the form', () => { + const goal = { + id: 'goal-1', + revision: 1, + sessionId: 'session-1', + condition: 'All tests pass', + status: 'active' as const, + setAt: 1, + iterations: 0, + maxIterations: 50, + consecutiveNoProgress: 0, + blockCap: 8, + tokensAtStart: 0, + tokensNow: 0, + tokensBaselinePending: false, + }; + const cases: Array<{ + outcome: GoalArmOutcome; + expected: ReturnType; + }> = [ + { + outcome: { kind: 'armed', goal }, + expected: { action: 'close' }, + }, + { + outcome: { + kind: 'reconciled', + currentGoal: goal, + matchesRequestedState: true, + }, + expected: { action: 'lock', notice: { kind: 'matching_goal', goal } }, + }, + { + outcome: { + kind: 'reconciled', + currentGoal: goal, + matchesRequestedState: false, + }, + expected: { action: 'lock', notice: { kind: 'different_goal', goal } }, + }, + { + outcome: { + kind: 'reconciled', + currentGoal: null, + matchesRequestedState: false, + }, + expected: { action: 'lock', notice: { kind: 'no_goal' } }, + }, + { + outcome: { kind: 'reconciliation_unavailable' }, + expected: { action: 'lock', notice: { kind: 'unavailable' } }, + }, + ]; + + for (const { outcome, expected } of cases) { + assert.deepEqual(interpretGoalArmOutcome(outcome), expected); + } +}); + +test('Goal reconciliation copy explains authoritative state in Chinese and English', () => { + const zh = getShellCopy('zh').goalDialog; + assert.match( + zh.reconciledMatching('所有测试通过', zh.statusLabels.active), + /所有测试通过.*进行中.*无法确认.*提交/, + ); + assert.match(zh.reconciledNoGoal, /未读到 Goal/); + assert.match(zh.reconciliationUnavailable, /不会重复提交/); + + const en = getShellCopy('en').goalDialog; + assert.match( + en.reconciledDifferent('All tests pass', en.statusLabels.paused), + /All tests pass.*Paused.*differs/, + ); + assert.match(en.reconciliationUnavailable, /will not submit twice/); +}); diff --git a/apps/desktop/src/main/__tests__/goal-dialog.test.tsx b/apps/desktop/src/main/__tests__/goal-dialog.test.tsx new file mode 100644 index 0000000000..cf588d3582 --- /dev/null +++ b/apps/desktop/src/main/__tests__/goal-dialog.test.tsx @@ -0,0 +1,251 @@ +import assert from 'node:assert/strict'; +import { afterEach, test } from 'node:test'; +import { act, createElement } from 'react'; +import { createRoot, type Root } from 'react-dom/client'; +import { parseHTML } from 'linkedom'; +import { AstryxLocaleProvider, LocaleProvider } from '@maka/ui'; +import type { GoalArmOutcome } from '../../shared/goal-arm.js'; +import { GoalDialog } from '../../renderer/goal-dialog.js'; + +const originalGlobals = { + document: globalThis.document, + window: globalThis.window, + HTMLElement: globalThis.HTMLElement, + HTMLIFrameElement: globalThis.HTMLIFrameElement, + Event: globalThis.Event, + Node: globalThis.Node, + CSS: globalThis.CSS, + matchMedia: globalThis.matchMedia, + requestAnimationFrame: globalThis.requestAnimationFrame, + cancelAnimationFrame: globalThis.cancelAnimationFrame, + IS_REACT_ACT_ENVIRONMENT: (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }) + .IS_REACT_ACT_ENVIRONMENT, +}; + +let mountedRoot: Root | undefined; + +afterEach(async () => { + if (mountedRoot) await act(() => mountedRoot?.unmount()); + mountedRoot = undefined; + Object.assign(globalThis, originalGlobals); +}); + +test('ignores a stale Goal arm result after switching Sessions', async () => { + const first = deferred(); + const calls: string[] = []; + const harness = installGoalDialog(async (sessionId) => { + calls.push(sessionId); + return first.promise; + }); + await harness.render('session-1'); + await setInputValue(harness.document, 'textarea', 'Finish session one'); + await clickButton(harness.document, 'Start'); + assert.deepEqual(calls, ['session-1']); + + await harness.render('session-2'); + first.resolve({ kind: 'reconciliation_unavailable' }); + await act(async () => { + await first.promise; + await Promise.resolve(); + }); + + assert.doesNotMatch(harness.document.body.textContent, /cannot be confirmed/); + assert.equal(harness.closed, 0); +}); + +test('ignores a stale Goal arm rejection after closing and reopening', async () => { + const first = deferred(); + const harness = installGoalDialog(async () => first.promise); + await harness.render('session-1'); + await setInputValue(harness.document, 'textarea', 'Finish session one'); + await clickButton(harness.document, 'Start'); + + await harness.render(undefined); + await harness.render('session-1'); + first.reject(new Error('old rejection')); + await act(async () => { + await first.promise.catch(() => undefined); + await Promise.resolve(); + }); + + assert.doesNotMatch(harness.document.body.textContent, /goal could not be set/i); + assert.equal(harness.document.querySelector('textarea')?.hasAttribute('disabled'), false); + assert.equal(harness.closed, 0); +}); + +test('closes only for armed and locks reconciled state until reopen', async () => { + const goal = goalState(); + let outcome: GoalArmOutcome = { + kind: 'reconciled', + currentGoal: null, + matchesRequestedState: false, + }; + const harness = installGoalDialog(async () => outcome); + await harness.render('session-1'); + await setInputValue(harness.document, 'textarea', 'Finish session one'); + await clickButton(harness.document, 'Start'); + + assert.equal(harness.closed, 0); + assert.match(harness.document.body.textContent, /no Goal was found/); + assert.equal(harness.document.querySelector('textarea')?.hasAttribute('disabled'), true); + assert.equal(findButton(harness.document, 'Start').hasAttribute('disabled'), true); + + await harness.render(undefined); + await harness.render('session-1'); + assert.doesNotMatch(harness.document.body.textContent, /no Goal was found/); + assert.equal(harness.document.querySelector('textarea')?.hasAttribute('disabled'), false); + + outcome = { kind: 'armed', goal }; + await setInputValue(harness.document, 'textarea', 'Finish session one'); + await clickButton(harness.document, 'Start'); + assert.equal(harness.closed, 1); +}); + +test('keeps the Goal form editable after a deterministic rejection', async () => { + const harness = installGoalDialog(async () => { + throw new Error('Goal already exists'); + }); + await harness.render('session-1'); + await setInputValue(harness.document, 'textarea', 'Finish session one'); + await clickButton(harness.document, 'Start'); + + assert.equal(harness.closed, 0); + assert.match(harness.document.body.textContent, /goal could not be set/i); + assert.equal(harness.document.querySelector('textarea')?.hasAttribute('disabled'), false); + assert.equal(findButton(harness.document, 'Start').hasAttribute('disabled'), false); +}); + +function installGoalDialog( + arm: (sessionId: string) => Promise, +) { + const parsed = parseHTML('
'); + const { document, window } = parsed; + const matchMedia = (media: string) => ({ + matches: false, + media, + onchange: null, + addListener() {}, + removeListener() {}, + addEventListener() {}, + removeEventListener() {}, + dispatchEvent: () => false, + }); + Object.assign(window, { matchMedia, scrollTo() {} }); + Object.assign(window.HTMLElement.prototype, { + showModal(this: HTMLElement) { + this.setAttribute('open', ''); + }, + close(this: HTMLElement) { + this.removeAttribute('open'); + }, + }); + Object.assign(globalThis, { + document, + window, + matchMedia, + HTMLElement: window.HTMLElement, + HTMLIFrameElement: window.HTMLIFrameElement ?? class HTMLIFrameElement {}, + Event: window.Event, + Node: window.Node, + CSS: { escape: (value: string) => value }, + requestAnimationFrame: (callback: FrameRequestCallback) => setTimeout(callback, 0), + cancelAnimationFrame: (handle: number) => clearTimeout(handle), + IS_REACT_ACT_ENVIRONMENT: true, + }); + (window as unknown as { maka: unknown }).maka = { + goal: { arm }, + }; + const container = document.querySelector('#root'); + assert.ok(container); + const root = createRoot(container); + mountedRoot = root; + let closed = 0; + return { + document, + get closed() { + return closed; + }, + async render(sessionId: string | undefined) { + await act(async () => { + root.render( + createElement(LocaleProvider, { + locale: 'en', + children: createElement(AstryxLocaleProvider, { + children: createElement(GoalDialog, { + sessionId, + onClose: () => { + closed += 1; + }, + }), + }), + }), + ); + await Promise.resolve(); + }); + }, + }; +} + +async function setInputValue( + document: Document, + selector: string, + value: string, +): Promise { + const input = document.querySelector(selector) as HTMLInputElement | null; + assert.ok(input, `missing input: ${selector}`); + await act(async () => { + input.value = value; + const propsKey = Object.keys(input).find((key) => key.startsWith('__reactProps$')); + assert.ok(propsKey, 'missing React props on input'); + const props = (input as unknown as Record)[propsKey] as { + onChange?: (event: { target: HTMLInputElement; defaultPrevented: boolean }) => void; + }; + assert.ok(props.onChange, 'missing React change handler'); + props.onChange({ target: input, defaultPrevented: false }); + await Promise.resolve(); + }); +} + +async function clickButton(document: Document, label: string): Promise { + const button = findButton(document, label); + await act(async () => { + button.click(); + await Promise.resolve(); + }); +} + +function findButton(document: Document, label: string): HTMLButtonElement { + const button = [...document.querySelectorAll('button')].find( + (candidate) => candidate.textContent === label, + ) as HTMLButtonElement | undefined; + assert.ok(button, `missing button: ${label}`); + return button; +} + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((accept, decline) => { + resolve = accept; + reject = decline; + }); + return { promise, resolve, reject }; +} + +function goalState() { + return { + id: 'goal-1', + revision: 1, + sessionId: 'session-1', + condition: 'Finish session one', + status: 'active' as const, + setAt: 1, + iterations: 0, + maxIterations: 50, + consecutiveNoProgress: 0, + blockCap: 8, + tokensAtStart: 0, + tokensNow: 0, + tokensBaselinePending: false, + }; +} diff --git a/apps/desktop/src/main/__tests__/goal-preload-bridge.test.ts b/apps/desktop/src/main/__tests__/goal-preload-bridge.test.ts new file mode 100644 index 0000000000..8597729599 --- /dev/null +++ b/apps/desktop/src/main/__tests__/goal-preload-bridge.test.ts @@ -0,0 +1,70 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + invokeProjectedSessionRuntimeHost, +} from '../../preload/projected-session-runtime-host.js'; +import { desktopSessionKey } from '../../shared/runtime-host-identity.js'; + +test('Goal arm preload routing preserves target scope and projects every outcome', async () => { + const scope = { hostId: 'host-1', targetEpoch: 'epoch-2' }; + const desktopSessionId = desktopSessionKey({ + hostId: scope.hostId, + sessionId: 'session-1', + }); + const request = { condition: 'Finish', maxIterations: 20, tokenBudget: 1_000 }; + const scenarios = [ + { + wire: { + kind: 'armed', + goal: { id: 'goal-1', sessionId: 'session-1' }, + }, + projected: { + kind: 'armed', + goal: { id: 'goal-1', sessionId: desktopSessionId }, + }, + }, + { + wire: { + kind: 'reconciled', + currentGoal: { id: 'goal-2', sessionId: 'session-1' }, + matchesRequestedState: true, + }, + projected: { + kind: 'reconciled', + currentGoal: { id: 'goal-2', sessionId: desktopSessionId }, + matchesRequestedState: true, + }, + }, + { + wire: { kind: 'reconciliation_unavailable' }, + projected: { kind: 'reconciliation_unavailable' }, + }, + ] as const; + + for (const scenario of scenarios) { + const invocations: unknown[] = []; + const result = await invokeProjectedSessionRuntimeHost( + async (sessionId) => { + assert.equal(sessionId, desktopSessionId); + return { scope, sessionId: 'session-1' }; + }, + async (channel, targetScope, rawSessionId, ...args) => { + invocations.push({ channel, targetScope, rawSessionId, args }); + return scenario.wire; + }, + 'goal:arm', + desktopSessionId, + request, + ); + + assert.deepEqual(invocations, [ + { + channel: 'goal:arm', + targetScope: scope, + rawSessionId: 'session-1', + args: [request], + }, + ]); + assert.deepEqual(result, scenario.projected); + } +}); diff --git a/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts index 7cb9dfb738..9d7fbdcc13 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts @@ -9,7 +9,212 @@ import { readWithFallback, tryReconnectableReadResult, } from "../ipc-reconnect-policy.js"; -import { RuntimeHostReconnectingIpcMain } from "../runtime-host-reconnecting-ipc-main.js"; +import * as ipcReconnectPolicy from "../ipc-reconnect-policy.js"; +import { + RuntimeHostReconnectingIpcMain, + RuntimeHostTargetChangedError, +} from "../runtime-host-reconnecting-ipc-main.js"; + +test("classifies only dispatched control connection loss for reconciliation", () => { + const predicate = ( + ipcReconnectPolicy as typeof ipcReconnectPolicy & { + isDispatchedControlConnectionLoss?: (error: unknown) => boolean; + } + ).isDispatchedControlConnectionLoss; + assert.equal(typeof predicate, "function"); + assert.equal( + predicate?.( + new RuntimeHostRequestInterruptedError( + "goal.arm", + "control", + "dispatched", + "connection_lost", + ), + ), + true, + ); + for (const error of [ + new RuntimeHostRequestInterruptedError( + "goal.arm", + "control", + "not_dispatched", + "connection_lost", + ), + new RuntimeHostRequestInterruptedError( + "goal.arm", + "control", + "dispatched", + "timeout", + ), + new RuntimeHostRequestInterruptedError( + "goal.query", + "query", + "dispatched", + "connection_lost", + ), + new RuntimeHostRequestInterruptedError( + "web-search.execute", + "command", + "dispatched", + "connection_lost", + ), + new Error("ordinary failure"), + ]) { + assert.equal(predicate?.(error), false); + } +}); + +test("reconciles a dispatched control on a replacement without replaying it", async () => { + const ipc = ipcHarness(); + const router = new RuntimeHostReconnectingIpcMain(ipc); + const firstTarget = router.createTarget("target-a") as ReconciledControlTarget; + assert.equal(typeof firstTarget.handleReconciledControl, "function"); + let dispatches = 0; + firstTarget.handleReconciledControl("goal:arm", { + dispatch: async () => { + dispatches += 1; + return { + kind: "reconcile", + context: { condition: "All tests pass" }, + }; + }, + reconcile: async () => assert.fail("The closed candidate must not reconcile"), + }); + router.activate("target-a"); + + const arming = ipc.invoke("goal:arm", scope("target-a")); + await new Promise((resolve) => setImmediate(resolve)); + firstTarget.removeHandler("goal:arm"); + const replacementTarget = router.createTarget("target-a") as ReconciledControlTarget; + let reconciliations = 0; + replacementTarget.handleReconciledControl("goal:arm", { + dispatch: async () => assert.fail("The mutation must not be replayed"), + reconcile: async (context) => { + reconciliations += 1; + assert.deepEqual(context, { condition: "All tests pass" }); + return { kind: "reconciled", currentGoal: "goal-1" }; + }, + }); + + assert.deepEqual(await arming, { + kind: "reconciled", + currentGoal: "goal-1", + }); + assert.equal(dispatches, 1); + assert.equal(reconciliations, 1); + router.close(); +}); + +test("retries only reconciliation when its replacement connection is lost", async () => { + const ipc = ipcHarness(); + const router = new RuntimeHostReconnectingIpcMain(ipc); + const firstTarget = router.createTarget("target-a") as ReconciledControlTarget; + let dispatches = 0; + firstTarget.handleReconciledControl("goal:arm", { + dispatch: async () => { + dispatches += 1; + return { kind: "reconcile", context: { sessionId: "session-1" } }; + }, + reconcile: async () => assert.fail("The closed candidate must not reconcile"), + }); + router.activate("target-a"); + + const arming = ipc.invoke("goal:arm", scope("target-a")); + const settled = arming.then( + (value) => ({ ok: true as const, value }), + (error: unknown) => ({ ok: false as const, error }), + ); + await new Promise((resolve) => setImmediate(resolve)); + firstTarget.removeHandler("goal:arm"); + const failedTarget = router.createTarget("target-a") as ReconciledControlTarget; + const reconciliationEntered = deferred(); + const failReconciliation = deferred(); + failedTarget.handleReconciledControl("goal:arm", { + dispatch: async () => assert.fail("The mutation must not be replayed"), + reconcile: async () => { + reconciliationEntered.resolve(); + await failReconciliation.promise; + throw new RuntimeHostRequestInterruptedError( + "goal.query", + "query", + "dispatched", + "connection_lost", + ); + }, + }); + await reconciliationEntered.promise; + failReconciliation.resolve(); + await new Promise((resolve) => setImmediate(resolve)); + failedTarget.removeHandler("goal:arm"); + const recoveredTarget = router.createTarget("target-a") as ReconciledControlTarget; + let reconciliations = 0; + recoveredTarget.handleReconciledControl("goal:arm", { + dispatch: async () => assert.fail("The mutation must not be replayed"), + reconcile: async (context) => { + reconciliations += 1; + assert.deepEqual(context, { sessionId: "session-1" }); + return { kind: "reconciled", currentGoal: "goal-1" }; + }, + }); + + assert.deepEqual(await settled, { + ok: true, + value: { kind: "reconciled", currentGoal: "goal-1" }, + }); + assert.equal(dispatches, 1); + assert.equal(reconciliations, 1); + router.close(); +}); + +test("never reconciles a control through a different target epoch", async () => { + const ipc = ipcHarness(); + const router = new RuntimeHostReconnectingIpcMain(ipc); + const firstTarget = router.createTarget("target-a") as ReconciledControlTarget; + let dispatches = 0; + firstTarget.handleReconciledControl("goal:arm", { + dispatch: async () => { + dispatches += 1; + return { kind: "reconcile", context: { sessionId: "session-1" } }; + }, + reconcile: async () => assert.fail("The closed candidate must not reconcile"), + }); + router.activate("target-a"); + + const arming = ipc.invoke("goal:arm", scope("target-a")); + const settled = arming.then( + () => ({ settled: true }), + (error: unknown) => ({ settled: true, error }), + ); + await new Promise((resolve) => setImmediate(resolve)); + firstTarget.removeHandler("goal:arm"); + const otherTarget = router.createTarget("target-b") as ReconciledControlTarget; + let otherReconciliations = 0; + otherTarget.handleReconciledControl("goal:arm", { + dispatch: async () => assert.fail("The mutation must not be replayed"), + reconcile: async () => { + otherReconciliations += 1; + return { kind: "reconciled" }; + }, + }); + router.activate("target-b"); + const pending = Promise.race([ + settled, + new Promise<{ settled: false }>((resolve) => + setImmediate(() => resolve({ settled: false })), + ), + ]); + assert.deepEqual(await pending, { settled: false }); + + router.deactivate("target-a"); + const result = await settled; + assert.equal(result.settled, true); + assert.ok( + "error" in result && result.error instanceof RuntimeHostTargetChangedError, + ); + assert.equal(dispatches, 1); + assert.equal(otherReconciliations, 0); + router.close(); +}); test("holds an invocation across a Runtime Host candidate replacement", async () => { const ipc = ipcHarness(); @@ -257,6 +462,18 @@ test("read adapters project ordinary failures without hiding reconnectable failu type IpcHandler = Parameters[1]; +type ReconciledControlTarget = ReturnType< + RuntimeHostReconnectingIpcMain["createTarget"] +> & { + handleReconciledControl( + channel: string, + handlers: { + dispatch: IpcHandler; + reconcile: (context: unknown, ...args: Parameters) => Promise; + }, + ): void; +}; + function ipcHarness() { const handlers = new Map(); return { diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-domains-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-domains-ipc-main.test.ts index 1a9fe9c4b1..912322a552 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-domains-ipc-main.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-domains-ipc-main.test.ts @@ -5,7 +5,18 @@ import { projectDeepResearchClientProgress } from '@maka/core/deep-research-clie import { type DeepResearchRun } from '@maka/core/deep-research-run'; import { type PlanSessionState } from '@maka/core/plan'; import { type ShellRunUpdate } from '@maka/core/events'; -import { encodeDeepResearchSnapshot } from '@maka/runtime-host/protocol'; +import { + encodeDeepResearchSnapshot, + type GoalProjection, +} from '@maka/runtime-host/protocol'; +import { + RuntimeHostOperationError, + RuntimeHostRequestInterruptedError, +} from '@maka/runtime-host/client'; +import type { + ReconciledControlHandlers, + ReconnectableReadIpcMain, +} from '../ipc-reconnect-policy.js'; import { projectEmbeddedDeepResearch, projectHostedDeepResearch, @@ -17,9 +28,336 @@ import { type DomainClient = RuntimeHostSessionDomainsIpcDeps['client']; +test('goal:arm reconciles a lost dispatched response without dispatching again', async () => { + const previous = goalProjection({ goalId: 'goal-old', condition: 'Old goal' }); + const requested = { + condition: ' Finish the adapter ', + maxIterations: 20, + tokenBudget: 1_000, + }; + let armCalls = 0; + const firstIpc = reconciledIpcHarness(); + registerDomainsIpc( + { + client: domainClient({ + queryGoal: async () => ({ sessionId: 'session-1', goal: previous }), + armGoal: async () => { + armCalls += 1; + throw new RuntimeHostRequestInterruptedError( + 'goal.arm', + 'control', + 'dispatched', + 'connection_lost', + ); + }, + }), + emitModeChanged() {}, + }, + firstIpc, + ); + + const step = await firstIpc.dispatch('goal:arm', 'session-1', requested); + assert.equal((step as { kind: string }).kind, 'reconcile'); + + const replacementIpc = reconciledIpcHarness(); + registerDomainsIpc( + { + client: domainClient({ + queryGoal: async () => ({ + sessionId: 'session-1', + goal: goalProjection({ condition: 'Finish the adapter' }), + }), + armGoal: async () => assert.fail('replacement must never re-arm the Goal'), + }), + emitModeChanged() {}, + }, + replacementIpc, + ); + + assert.deepEqual( + await replacementIpc.reconcile( + 'goal:arm', + (step as { context: unknown }).context, + 'session-1', + requested, + ), + { + kind: 'reconciled', + currentGoal: { + id: 'goal-1', + revision: 3, + sessionId: 'session-1', + condition: 'Finish the adapter', + status: 'active', + setAt: 1, + iterations: 2, + maxIterations: 20, + consecutiveNoProgress: 0, + blockCap: 8, + tokenBudget: 1_000, + tokensAtStart: 0, + tokensNow: 120, + tokensBaselinePending: false, + }, + matchesRequestedState: true, + }, + ); + assert.equal(armCalls, 1); +}); + +test('goal:arm reconciliation compares every canonical requested field', async () => { + const firstIpc = reconciledIpcHarness(); + registerDomainsIpc( + { + client: domainClient({ + queryGoal: async () => ({ + sessionId: 'session-1', + goal: goalProjection({ goalId: 'goal-old', condition: 'Old goal' }), + }), + armGoal: async () => { + throw new RuntimeHostRequestInterruptedError( + 'goal.arm', + 'control', + 'dispatched', + 'connection_lost', + ); + }, + }), + emitModeChanged() {}, + }, + firstIpc, + ); + const requested = { condition: ' Default budget goal ' }; + const step = await firstIpc.dispatch('goal:arm', 'session-1', requested); + const context = (step as { context: unknown }).context; + const cases: ReadonlyArray<{ + readonly name: string; + readonly goal: GoalProjection; + readonly expected: boolean; + }> = [ + { + name: 'a new Goal matches the trimmed condition and default iteration budget', + goal: goalProjection({ + goalId: 'goal-new', + condition: 'Default budget goal', + maxIterations: 50, + tokenBudget: null, + }), + expected: true, + }, + { + name: 'the previous Goal identity does not match', + goal: goalProjection({ + goalId: 'goal-old', + condition: 'Default budget goal', + maxIterations: 50, + tokenBudget: null, + }), + expected: false, + }, + { + name: 'a different Session does not match', + goal: goalProjection({ + goalId: 'goal-new', + sessionId: 'session-2', + condition: 'Default budget goal', + maxIterations: 50, + tokenBudget: null, + }), + expected: false, + }, + { + name: 'a different condition does not match', + goal: goalProjection({ + goalId: 'goal-new', + condition: 'Different goal', + maxIterations: 50, + tokenBudget: null, + }), + expected: false, + }, + { + name: 'a different iteration budget does not match', + goal: goalProjection({ + goalId: 'goal-new', + condition: 'Default budget goal', + maxIterations: 49, + tokenBudget: null, + }), + expected: false, + }, + { + name: 'a different token budget does not match', + goal: goalProjection({ + goalId: 'goal-new', + condition: 'Default budget goal', + maxIterations: 50, + tokenBudget: 1, + }), + expected: false, + }, + ]; + + for (const scenario of cases) { + const ipc = reconciledIpcHarness(); + registerDomainsIpc( + { + client: domainClient({ + queryGoal: async () => ({ sessionId: 'session-1', goal: scenario.goal }), + }), + emitModeChanged() {}, + }, + ipc, + ); + const outcome = await ipc.reconcile('goal:arm', context, 'session-1', requested) as { + readonly matchesRequestedState: boolean; + }; + assert.equal(outcome.matchesRequestedState, scenario.expected, scenario.name); + } +}); + +test('goal:arm reconciliation reports different, missing, and unavailable authority', async () => { + const firstIpc = reconciledIpcHarness(); + registerDomainsIpc( + { + client: domainClient({ + queryGoal: async () => ({ + sessionId: 'session-1', + goal: goalProjection({ goalId: 'goal-old', condition: 'Old goal' }), + }), + armGoal: async () => { + throw new RuntimeHostRequestInterruptedError( + 'goal.arm', + 'control', + 'dispatched', + 'connection_lost', + ); + }, + }), + emitModeChanged() {}, + }, + firstIpc, + ); + const step = await firstIpc.dispatch('goal:arm', 'session-1', { + condition: 'Default budget goal', + }); + const context = (step as { context: unknown }).context; + + const reconcileWith = async ( + queryGoal: DomainClient['queryGoal'], + ): Promise => { + const ipc = reconciledIpcHarness(); + registerDomainsIpc( + { + client: domainClient({ queryGoal }), + emitModeChanged() {}, + }, + ipc, + ); + return ipc.reconcile('goal:arm', context, 'session-1', { + condition: 'Default budget goal', + }); + }; + + const oldGoal = goalProjection({ + goalId: 'goal-old', + condition: 'Default budget goal', + maxIterations: 50, + tokenBudget: null, + }); + assert.deepEqual( + await reconcileWith(async () => ({ sessionId: 'session-1', goal: oldGoal })), + { + kind: 'reconciled', + currentGoal: { + id: 'goal-old', + revision: 3, + sessionId: 'session-1', + condition: 'Default budget goal', + status: 'active', + setAt: 1, + iterations: 2, + maxIterations: 50, + consecutiveNoProgress: 0, + blockCap: 8, + tokensAtStart: 0, + tokensNow: 120, + tokensBaselinePending: false, + }, + matchesRequestedState: false, + }, + ); + assert.deepEqual( + await reconcileWith(async () => ({ sessionId: 'session-1', goal: null })), + { kind: 'reconciled', currentGoal: null, matchesRequestedState: false }, + ); + assert.deepEqual( + await reconcileWith(async () => { + throw new Error('query unavailable'); + }), + { kind: 'reconciliation_unavailable' }, + ); + await assert.rejects( + reconcileWith(async () => { + throw new RuntimeHostRequestInterruptedError( + 'goal.query', + 'query', + 'dispatched', + 'connection_lost', + ); + }), + RuntimeHostRequestInterruptedError, + ); +}); + +test('goal:arm preserves deterministic and non-dispatched rejection semantics', async () => { + const errors = [ + new RuntimeHostRequestInterruptedError( + 'goal.arm', + 'control', + 'not_dispatched', + 'connection_lost', + ), + new RuntimeHostRequestInterruptedError( + 'goal.arm', + 'control', + 'dispatched', + 'timeout', + ), + new RuntimeHostOperationError( + 'goal.arm', + 'host_draining', + 'Runtime Host is draining', + ), + ]; + for (const expected of errors) { + const ipc = ipcHarness(); + let armCalls = 0; + registerDomainsIpc( + { + client: domainClient({ + queryGoal: async () => ({ sessionId: 'session-1', goal: null }), + armGoal: async () => { + armCalls += 1; + throw expected; + }, + }), + emitModeChanged() {}, + }, + ipc, + ); + await assert.rejects( + ipc.invoke('goal:arm', 'session-1', { condition: 'Finish' }), + (actual) => actual === expected, + ); + assert.equal(armCalls, 1); + } +}); + test('goal:arm takes the Session from the scoped channel and refuses any other key', async () => { const armed: unknown[] = []; const client = domainClient({ + queryGoal: async (sessionId) => ({ sessionId, goal: null }), armGoal: async (input) => { armed.push(input); return { sessionId: input.sessionId, goal: goalProjection() }; @@ -28,7 +366,7 @@ test('goal:arm takes the Session from the scoped channel and refuses any other k const ipc = ipcHarness(); registerDomainsIpc({ client, emitModeChanged() {} }, ipc); - const goal = await ipc.invoke('goal:arm', 'session-1', { + const outcome = await ipc.invoke('goal:arm', 'session-1', { condition: 'Finish the adapter', maxIterations: 20, tokenBudget: 1_000, @@ -41,7 +379,14 @@ test('goal:arm takes the Session from the scoped channel and refuses any other k tokenBudget: 1_000, }, ]); - assert.equal((goal as { id: string }).id, 'goal-1'); + assert.equal( + (outcome as { kind: string; goal: { id: string } }).kind, + 'armed', + ); + assert.equal( + (outcome as { kind: string; goal: { id: string } }).goal.id, + 'goal-1', + ); // Omitted budgets are "not chosen", which the Host reads as its defaults. await ipc.invoke('goal:arm', 'session-1', { condition: 'Finish the adapter' }); @@ -88,6 +433,9 @@ test('adapts Host Goal, Task, Deep Research, and Resource projections', async () clearGoal: async (sessionId) => { controls.push(sessionId); }, + controlGoalWithRetry: async (sessionId, action) => { + controls.push({ sessionId, action }); + }, queryDeepResearch: async () => hostedResearch(), }); const ipc = ipcHarness(); @@ -116,7 +464,13 @@ test('adapts Host Goal, Task, Deep Research, and Resource projections', async () tokensBaselinePending: false, }); await ipc.invoke('goal:clear', 'session-1'); - assert.deepEqual(controls, ['session-1']); + await ipc.invoke('goal:pause', 'session-1'); + await ipc.invoke('goal:resume', 'session-1'); + assert.deepEqual(controls, [ + 'session-1', + { sessionId: 'session-1', action: 'pause' }, + { sessionId: 'session-1', action: 'resume' }, + ]); assert.deepEqual(await ipc.invoke('deepResearch:get', 'session-1'), { sessionId: 'session-1', objective: 'Inspect the adapter', @@ -781,6 +1135,7 @@ function domainClient(overrides: Partial): DomainClient { return { armGoal: unavailable, clearGoal: unavailable, + controlGoalWithRetry: unavailable, acquireRuntimeResourceController: unavailable, controlPlan: unavailable, controlRuntimeResource: unavailable, @@ -802,7 +1157,13 @@ function domainClient(overrides: Partial): DomainClient { } as DomainClient; } -function goalProjection() { +function goalProjection( + overrides: Partial = {}, +): GoalProjection { + return { ...baseGoalProjection(), ...overrides }; +} + +function baseGoalProjection() { return { goalId: 'goal-1', revision: 3, @@ -938,10 +1299,49 @@ function ipcHarness() { }; } +function reconciledIpcHarness() { + const ordinaryHandlers = new Map(); + const controlHandlers = new Map< + string, + ReconciledControlHandlers + >(); + return { + handle(channel: string, handler: IpcHandler) { + ordinaryHandlers.set(channel, handler); + }, + handleReconciledControl( + channel: string, + handlers: ReconciledControlHandlers, + ) { + controlHandlers.set( + channel, + handlers as unknown as ReconciledControlHandlers, + ); + }, + async dispatch(channel: string, ...args: unknown[]): Promise { + const handlers = controlHandlers.get(channel); + assert.ok(handlers, `missing reconciled control: ${channel}`); + return handlers.dispatch({} as never, ...args); + }, + async reconcile( + channel: string, + context: unknown, + ...args: unknown[] + ): Promise { + const handlers = controlHandlers.get(channel); + assert.ok(handlers, `missing reconciled control: ${channel}`); + return handlers.reconcile(context, {} as never, ...args); + }, + } satisfies ReconnectableReadIpcMain & { + dispatch(channel: string, ...args: unknown[]): Promise; + reconcile(channel: string, context: unknown, ...args: unknown[]): Promise; + }; +} + function registerDomainsIpc( deps: Omit & Partial>, - ipcMain: Pick, + ipcMain: ReconnectableReadIpcMain, ) { return registerRuntimeHostSessionDomainsIpc( { diff --git a/apps/desktop/src/main/ipc-reconnect-policy.ts b/apps/desktop/src/main/ipc-reconnect-policy.ts index a25129f963..a1e5d61a98 100644 --- a/apps/desktop/src/main/ipc-reconnect-policy.ts +++ b/apps/desktop/src/main/ipc-reconnect-policy.ts @@ -8,8 +8,28 @@ import { HOST_OPERATION_SPECS } from "@maka/runtime-host/protocol"; export type IpcHandler = Parameters[1]; +export type ReconciledControlStep = + | { readonly kind: "completed"; readonly value: Result } + | { readonly kind: "reconcile"; readonly context: Context }; + +export interface ReconciledControlHandlers { + dispatch( + event: Parameters[0], + ...args: unknown[] + ): Promise>; + reconcile( + context: Context, + event: Parameters[0], + ...args: unknown[] + ): Promise; +} + export interface ReconnectableReadIpcMain extends Pick { handleReconnectableRead?(channel: string, listener: IpcHandler): void; + handleReconciledControl?( + channel: string, + handlers: ReconciledControlHandlers, + ): void; } export function handleReconnectableRead( @@ -24,6 +44,23 @@ export function handleReconnectableRead( } } +export function handleReconciledControl( + ipcMain: ReconnectableReadIpcMain, + channel: string, + handlers: ReconciledControlHandlers, +): void { + if (ipcMain.handleReconciledControl) { + ipcMain.handleReconciledControl(channel, handlers); + return; + } + ipcMain.handle(channel, async (event, ...args) => { + const step = await handlers.dispatch(event, ...args); + return step.kind === "completed" + ? step.value + : handlers.reconcile(step.context, event, ...args); + }); +} + export function isReconnectableReadFailure(error: unknown): boolean { return ( (error instanceof RuntimeHostOperationError && @@ -35,6 +72,15 @@ export function isReconnectableReadFailure(error: unknown): boolean { ); } +export function isDispatchedControlConnectionLoss(error: unknown): boolean { + return ( + error instanceof RuntimeHostRequestInterruptedError && + error.mode === "control" && + error.dispatch === "dispatched" && + error.reason === "connection_lost" + ); +} + export function rethrowReconnectableReadFailure(error: unknown): void { if (isReconnectableReadFailure(error)) throw error; } diff --git a/apps/desktop/src/main/runtime-host-desktop-candidate.ts b/apps/desktop/src/main/runtime-host-desktop-candidate.ts index b6b3069fe7..eadfcb07d3 100644 --- a/apps/desktop/src/main/runtime-host-desktop-candidate.ts +++ b/apps/desktop/src/main/runtime-host-desktop-candidate.ts @@ -50,7 +50,11 @@ import { } from "./runtime-host-session-execution-ipc-main.js"; import { RuntimeHostSessionObservationRegistry } from "./runtime-host-session-observation-registry.js"; import { RuntimeHostSessionObserver } from "./runtime-host-session-observer.js"; -import type { IpcHandler, ReconnectableReadIpcMain } from "./ipc-reconnect-policy.js"; +import type { + IpcHandler, + ReconciledControlHandlers, + ReconnectableReadIpcMain, +} from "./ipc-reconnect-policy.js"; import type { RuntimeHostTargetIpcMain } from "./runtime-host-reconnecting-ipc-main.js"; import { desktopSessionResourceKey, @@ -728,6 +732,41 @@ class ScopedIpcMain implements ReconnectableReadIpcMain { this.#handle(channel, listener, true); } + handleReconciledControl( + channel: string, + handlers: ReconciledControlHandlers, + ): void { + if (this.#closed) + throw new Error("Desktop Runtime Host candidate IPC is closed"); + if (this.#channels.has(channel)) { + throw new Error( + `Desktop Runtime Host candidate registered duplicate IPC: ${channel}`, + ); + } + const scopedHandlers: ReconciledControlHandlers = { + dispatch: (event, scope, ...args) => { + requireDesktopTargetScope(scope, this.scope); + return handlers.dispatch(event, ...args); + }, + reconcile: (context, event, scope, ...args) => { + requireDesktopTargetScope(scope, this.scope); + return handlers.reconcile(context, event, ...args); + }, + }; + if (this.#ipcMain.handleReconciledControl) { + this.#ipcMain.handleReconciledControl(channel, scopedHandlers); + } else { + this.#ipcMain.handle(channel, async (event, scope, ...args) => { + requireDesktopTargetScope(scope, this.scope); + const step = await handlers.dispatch(event, ...args); + return step.kind === "completed" + ? step.value + : handlers.reconcile(step.context, event, ...args); + }); + } + this.#channels.add(channel); + } + #handle(channel: string, listener: IpcHandler, reconnectableRead: boolean): void { if (this.#closed) throw new Error("Desktop Runtime Host candidate IPC is closed"); diff --git a/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts b/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts index c2e125a7d7..1ee2162c47 100644 --- a/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts @@ -2,9 +2,16 @@ import type { IpcMain } from "electron"; import { isReconnectableReadFailure, type IpcHandler, + type ReconciledControlHandlers, type ReconnectableReadIpcMain, } from "./ipc-reconnect-policy.js"; +type ReconcileIpcHandler = ( + context: unknown, + event: Parameters[0], + ...args: unknown[] +) => Promise; + interface HandlerWaiter { readonly epoch: string; readonly resolve: (handler: BoundHandler) => void; @@ -15,11 +22,13 @@ interface BoundHandler { readonly epoch: string; readonly owner: symbol; readonly listener: IpcHandler; + readonly reconcile?: ReconcileIpcHandler; } interface HandlerSlot { readonly waiters: Set; readonly reconnectableRead: boolean; + readonly reconciledControl: boolean; readonly handlers: Map; } @@ -60,9 +69,11 @@ export class RuntimeHostReconnectingIpcMain { epoch, isActive: () => this.#activeEpochs.has(epoch), handle: (channel, listener) => - this.#handle(epoch, owner, channel, listener, false), + this.#handle(epoch, owner, channel, listener, false, false), handleReconnectableRead: (channel, listener) => - this.#handle(epoch, owner, channel, listener, true), + this.#handle(epoch, owner, channel, listener, true, false), + handleReconciledControl: (channel, handlers) => + this.#handleReconciledControl(epoch, owner, channel, handlers), removeHandler: (channel) => this.#removeHandler(owner, channel), }; } @@ -101,6 +112,8 @@ export class RuntimeHostReconnectingIpcMain { channel: string, listener: IpcHandler, reconnectableRead: boolean, + reconciledControl: boolean, + reconcile?: ReconcileIpcHandler, ): void { if (this.#closed) throw new Error("Desktop Runtime Host IPC router is closed"); let slot = this.#slots.get(channel); @@ -109,6 +122,7 @@ export class RuntimeHostReconnectingIpcMain { handlers: new Map(), waiters: new Set(), reconnectableRead, + reconciledControl, }; this.#ipcMain.handle(channel, (event, ...args) => this.#dispatch(created, event, args), @@ -116,13 +130,16 @@ export class RuntimeHostReconnectingIpcMain { this.#slots.set(channel, created); slot = created; } - if (slot.reconnectableRead !== reconnectableRead) { + if ( + slot.reconnectableRead !== reconnectableRead || + slot.reconciledControl !== reconciledControl + ) { throw new Error(`Desktop Runtime Host IPC policy changed: ${channel}`); } if (slot.handlers.has(epoch)) { throw new Error(`Desktop Runtime Host IPC handler already exists: ${channel}`); } - const handler = { epoch, owner, listener }; + const handler = { epoch, owner, listener, ...(reconcile ? { reconcile } : {}) }; slot.handlers.set(epoch, handler); for (const waiter of [...slot.waiters]) { if (waiter.epoch !== epoch) continue; @@ -131,6 +148,23 @@ export class RuntimeHostReconnectingIpcMain { } } + #handleReconciledControl( + epoch: string, + owner: symbol, + channel: string, + handlers: ReconciledControlHandlers, + ): void { + this.#handle( + epoch, + owner, + channel, + handlers.dispatch as IpcHandler, + false, + true, + handlers.reconcile as unknown as ReconcileIpcHandler, + ); + } + #removeHandler(owner: symbol, channel: string): void { const slot = this.#slots.get(channel); if (!slot) return; @@ -147,22 +181,43 @@ export class RuntimeHostReconnectingIpcMain { const epoch = this.#requireTargetEpoch(args[0]); let handler = slot.handlers.get(epoch); if (!handler) handler = await this.#waitForHandler(slot, epoch); + let reconciliationContext: unknown; + let reconciling = false; while (true) { try { - const result = await handler.listener(event, ...args); + const result = reconciling + ? await requireReconcileHandler(handler)(reconciliationContext, event, ...args) + : await handler.listener(event, ...args); this.#assertActive(epoch); - if (slot.reconnectableRead && slot.handlers.get(epoch) !== handler) { + if ( + (slot.reconnectableRead || reconciling) && + slot.handlers.get(epoch) !== handler + ) { + handler = await this.#waitForHandler(slot, epoch, handler); + continue; + } + if (slot.reconciledControl && !reconciling) { + const step = requireReconciledControlStep(result); + if (step.kind === "completed") return step.value; + reconciliationContext = step.context; + reconciling = true; handler = await this.#waitForHandler(slot, epoch, handler); continue; } return result; } catch (error) { this.#assertActive(epoch); - if (slot.reconnectableRead && slot.handlers.get(epoch) !== handler) { + if ( + (slot.reconnectableRead || reconciling) && + slot.handlers.get(epoch) !== handler + ) { handler = await this.#waitForHandler(slot, epoch, handler); continue; } - if (!slot.reconnectableRead || !isReconnectableReadFailure(error)) { + if ( + (!slot.reconnectableRead && !reconciling) || + !isReconnectableReadFailure(error) + ) { throw error; } handler = await this.#waitForHandler(slot, epoch, handler); @@ -219,3 +274,25 @@ export class RuntimeHostReconnectingIpcMain { } } } + +function requireReconcileHandler(handler: BoundHandler): ReconcileIpcHandler { + if (!handler.reconcile) { + throw new Error("Desktop Runtime Host reconciled control handler is unavailable"); + } + return handler.reconcile; +} + +function requireReconciledControlStep( + value: unknown, +): { readonly kind: "completed"; readonly value: unknown } | { + readonly kind: "reconcile"; + readonly context: unknown; +} { + if (!value || typeof value !== "object") { + throw new Error("Desktop Runtime Host reconciled control returned an invalid step"); + } + const step = value as { kind?: unknown; value?: unknown; context?: unknown }; + if (step.kind === "completed") return { kind: "completed", value: step.value }; + if (step.kind === "reconcile") return { kind: "reconcile", context: step.context }; + throw new Error("Desktop Runtime Host reconciled control returned an invalid step"); +} diff --git a/apps/desktop/src/main/runtime-host-session-domains-ipc-main.ts b/apps/desktop/src/main/runtime-host-session-domains-ipc-main.ts index 5f8f0dc5b7..7cb2b389b4 100644 --- a/apps/desktop/src/main/runtime-host-session-domains-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-session-domains-ipc-main.ts @@ -5,7 +5,7 @@ import type { AgentGraphClientSnapshotOptions, AgentGraphOperatorInspection, } from '@maka/runtime/stream-graph-read-model'; -import type { GoalState } from '@maka/runtime/goal-state'; +import { DEFAULT_MAX_ITERATIONS, type GoalState } from '@maka/runtime/goal-state'; import type { ShellRunPtyDataEvent } from '@maka/runtime/shell-run-contract'; import type { GoalProjection, @@ -14,10 +14,16 @@ import type { import type { AgentGraphEpochDirectory } from '@maka/runtime-host/client'; import type { DesktopRuntimeHostClient } from './runtime-host-client.js'; import type { RuntimeHostSessionObserver } from './runtime-host-session-observer.js'; -import { GOAL_ARM_REQUEST_KEYS } from '../shared/goal-arm.js'; +import { + GOAL_ARM_REQUEST_KEYS, + type GoalArmOutcome, +} from '../shared/goal-arm.js'; import { projectHostedDeepResearch } from './deep-research-desktop-projection.js'; import { + handleReconciledControl, handleReconnectableRead, + isDispatchedControlConnectionLoss, + rethrowReconnectableReadFailure, type ReconnectableReadIpcMain, } from './ipc-reconnect-policy.js'; import { @@ -102,13 +108,50 @@ export function registerRuntimeHostSessionDomainsIpc( ipcMain.handle('goal:resume', async (_event, sessionId: unknown) => { await deps.client.controlGoalWithRetry(requiredId(sessionId, 'Session'), 'resume'); }); - ipcMain.handle('goal:arm', async (_event, sessionId: unknown, input: unknown) => { - const result = await deps.client.armGoal({ - sessionId: requiredId(sessionId, 'Session'), - ...requireGoalArmBudgets(input), - }); - return toDesktopGoal(result.goal); - }); + handleReconciledControl( + ipcMain, + 'goal:arm', + { + dispatch: async (_event, ...args) => { + const request = canonicalGoalArmRequest(args[0], args[1]); + const previous = await deps.client.queryGoal(request.sessionId); + try { + const result = await deps.client.armGoal(request); + return { + kind: 'completed', + value: { kind: 'armed', goal: toDesktopGoal(result.goal) }, + }; + } catch (error) { + if (!isDispatchedControlConnectionLoss(error)) throw error; + return { + kind: 'reconcile', + context: Object.freeze({ + request, + previousGoal: + previous.goal === null ? null : Object.freeze({ ...previous.goal }), + }), + }; + } + }, + reconcile: async (context) => { + try { + const result = await deps.client.queryGoal(context.request.sessionId); + return { + kind: 'reconciled', + currentGoal: result.goal === null ? null : toDesktopGoal(result.goal), + matchesRequestedState: goalMatchesArmRequest( + result.goal, + context.request, + context.previousGoal, + ), + }; + } catch (error) { + rethrowReconnectableReadFailure(error); + return { kind: 'reconciliation_unavailable' }; + } + }, + }, + ); handleReconnectableRead(ipcMain, 'plan-mode:getState', (_event, sessionId: unknown) => deps.client.getPlanState(requiredId(sessionId, 'Session')), @@ -311,6 +354,46 @@ async function refreshRuntimeResources( } } +interface CanonicalGoalArmRequest { + readonly sessionId: string; + readonly condition: string; + readonly maxIterations: number | null; + readonly tokenBudget: number | null; +} + +interface GoalArmReconciliationContext { + readonly request: CanonicalGoalArmRequest; + readonly previousGoal: GoalProjection | null; +} + +function canonicalGoalArmRequest( + sessionId: unknown, + input: unknown, +): CanonicalGoalArmRequest { + const budgets = requireGoalArmBudgets(input); + return Object.freeze({ + sessionId: requiredId(sessionId, 'Session'), + condition: budgets.condition.trim(), + maxIterations: budgets.maxIterations, + tokenBudget: budgets.tokenBudget, + }); +} + +function goalMatchesArmRequest( + current: GoalProjection | null, + request: CanonicalGoalArmRequest, + previous: GoalProjection | null, +): boolean { + return ( + current !== null && + current.goalId !== previous?.goalId && + current.sessionId === request.sessionId && + current.condition === request.condition && + current.maxIterations === (request.maxIterations ?? DEFAULT_MAX_ITERATIONS) && + current.tokenBudget === request.tokenBudget + ); +} + /** * The renderer sends what the user typed; the Host owns every bound. This only * gets the frame into the shape the protocol decodes — numbers stay numbers, diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index 8f01896c7a..ef0a74c86f 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -732,7 +732,7 @@ export interface MakaBridge { arm( sessionId: string, goal: import('../shared/goal-arm').GoalArmRequest, - ): Promise; + ): Promise; /** Clear the active goal, stopping autonomous continuation. */ clear(sessionId: string): Promise; /** Pause the active goal without spending a model turn. */ diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts index 62866f9470..e8569ab117 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -170,7 +170,11 @@ import { requireDesktopTargetScope, type DesktopTargetScope, } from '../shared/runtime-host-identity.js'; -import type { GoalArmRequest } from '../shared/goal-arm.js'; +import type { GoalArmOutcome, GoalArmRequest } from '../shared/goal-arm.js'; +import { + invokeProjectedSessionRuntimeHost as invokeProjectedSessionRuntimeHostBridge, + projectProtocolSessionIds, +} from './projected-session-runtime-host.js'; import { projectDesktopAttachmentRefs, projectDesktopDailyReviewSummary, @@ -450,14 +454,18 @@ async function invokeProjectedSessionRuntimeHost( sessionId: string, ...args: unknown[] ): Promise { - const session = await runtimeHostSessionRef(sessionId); - const value = await ipcRenderer.invoke( + return invokeProjectedSessionRuntimeHostBridge( + runtimeHostSessionRef, + (targetChannel, scope, rawSessionId, ...targetArgs) => ipcRenderer.invoke( + targetChannel, + scope, + rawSessionId, + ...targetArgs, + ), channel, - session.scope, - session.sessionId, + sessionId, ...args, - ) as T; - return projectProtocolSessionIds(session.scope.hostId, value); + ); } async function invokeSessionSummary( @@ -576,31 +584,6 @@ function projectShellRunUpdate( }; } -const SESSION_ID_FIELDS = new Set([ - 'sessionId', - 'rootSessionId', - 'childSessionId', - 'sourceSessionId', - 'ownerSessionId', -]); - -// Closed client models may be projected structurally. Opaque tool/provider data -// must use the typed projection above so user content is never rewritten. -function projectProtocolSessionIds(hostId: string, value: T): T { - if (Array.isArray(value)) { - return value.map((entry) => projectProtocolSessionIds(hostId, entry)) as T; - } - if (!value || typeof value !== 'object') return value; - return Object.fromEntries( - Object.entries(value).map(([key, entry]) => [ - key, - SESSION_ID_FIELDS.has(key) && typeof entry === 'string' - ? desktopSessionKey({ hostId, sessionId: entry }) - : projectProtocolSessionIds(hostId, entry), - ]), - ) as T; -} - function subscribeRuntimeHostEvent( channel: string, scope: DesktopTargetScope, @@ -1911,7 +1894,7 @@ const makaBridge = { get(sessionId: string): Promise { return invokeProjectedSessionRuntimeHost('goal:get', sessionId); }, - arm(sessionId: string, goal: GoalArmRequest): Promise { + arm(sessionId: string, goal: GoalArmRequest): Promise { return invokeProjectedSessionRuntimeHost('goal:arm', sessionId, goal); }, clear(sessionId: string): Promise { diff --git a/apps/desktop/src/preload/projected-session-runtime-host.ts b/apps/desktop/src/preload/projected-session-runtime-host.ts new file mode 100644 index 0000000000..967d232158 --- /dev/null +++ b/apps/desktop/src/preload/projected-session-runtime-host.ts @@ -0,0 +1,57 @@ +import { + desktopSessionKey, + type DesktopTargetScope, +} from '../shared/runtime-host-identity.js'; + +export interface ResolvedRuntimeHostSession { + readonly scope: DesktopTargetScope; + readonly sessionId: string; +} + +export type RuntimeHostSessionResolver = ( + sessionId: string, +) => Promise; + +export type ScopedSessionInvoker = ( + channel: string, + scope: DesktopTargetScope, + sessionId: string, + ...args: unknown[] +) => Promise; + +const SESSION_ID_FIELDS = new Set([ + 'sessionId', + 'rootSessionId', + 'childSessionId', + 'sourceSessionId', + 'ownerSessionId', +]); + +// Closed client models may be projected structurally. Opaque tool/provider data +// must use a typed projection so user content is never rewritten. +export function projectProtocolSessionIds(hostId: string, value: T): T { + if (Array.isArray(value)) { + return value.map((entry) => projectProtocolSessionIds(hostId, entry)) as T; + } + if (!value || typeof value !== 'object') return value; + return Object.fromEntries( + Object.entries(value).map(([key, entry]) => [ + key, + SESSION_ID_FIELDS.has(key) && typeof entry === 'string' + ? desktopSessionKey({ hostId, sessionId: entry }) + : projectProtocolSessionIds(hostId, entry), + ]), + ) as T; +} + +export async function invokeProjectedSessionRuntimeHost( + resolveSession: RuntimeHostSessionResolver, + invoke: ScopedSessionInvoker, + channel: string, + sessionId: string, + ...args: unknown[] +): Promise { + const session = await resolveSession(sessionId); + const value = await invoke(channel, session.scope, session.sessionId, ...args) as T; + return projectProtocolSessionIds(session.scope.hostId, value); +} diff --git a/apps/desktop/src/renderer/goal-arm-outcome.ts b/apps/desktop/src/renderer/goal-arm-outcome.ts new file mode 100644 index 0000000000..086a9355c3 --- /dev/null +++ b/apps/desktop/src/renderer/goal-arm-outcome.ts @@ -0,0 +1,31 @@ +import type { GoalState } from '@maka/runtime/goal-state'; +import type { GoalArmOutcome } from '../shared/goal-arm.js'; + +export type GoalArmReconciliationNotice = + | { readonly kind: 'matching_goal'; readonly goal: GoalState } + | { readonly kind: 'different_goal'; readonly goal: GoalState } + | { readonly kind: 'no_goal' } + | { readonly kind: 'unavailable' }; + +export type GoalArmOutcomeAction = + | { readonly action: 'close' } + | { readonly action: 'lock'; readonly notice: GoalArmReconciliationNotice }; + +export function interpretGoalArmOutcome( + outcome: GoalArmOutcome, +): GoalArmOutcomeAction { + if (outcome.kind === 'armed') return { action: 'close' }; + if (outcome.kind === 'reconciliation_unavailable') { + return { action: 'lock', notice: { kind: 'unavailable' } }; + } + if (outcome.currentGoal === null) { + return { action: 'lock', notice: { kind: 'no_goal' } }; + } + return { + action: 'lock', + notice: { + kind: outcome.matchesRequestedState ? 'matching_goal' : 'different_goal', + goal: outcome.currentGoal, + }, + }; +} diff --git a/apps/desktop/src/renderer/goal-dialog.tsx b/apps/desktop/src/renderer/goal-dialog.tsx index 3eeecc4fe1..41f1815f77 100644 --- a/apps/desktop/src/renderer/goal-dialog.tsx +++ b/apps/desktop/src/renderer/goal-dialog.tsx @@ -14,7 +14,7 @@ * budget the form is no longer showing, and a field that quietly keeps nothing * can arm no budget at all where the user asked for a tight one. */ -import { useEffect, useState } from 'react'; +import { useLayoutEffect, useRef, useState } from 'react'; import { Button } from '@astryxdesign/core/Button'; import { Dialog, DialogHeader } from '@astryxdesign/core/Dialog'; import { Layout, LayoutContent, LayoutFooter } from '@astryxdesign/core/Layout'; @@ -29,6 +29,10 @@ import { GOAL_TOKEN_BUDGET_MINIMUM, } from '@maka/core/goal'; import { getShellCopy, localizedShellErrorMessage } from './locales/shell-copy.js'; +import { + interpretGoalArmOutcome, + type GoalArmReconciliationNotice, +} from './goal-arm-outcome.js'; type BudgetReading = | { readonly kind: 'empty' } @@ -65,42 +69,77 @@ export function GoalDialog(props: { const [tokenBudgetText, setTokenBudgetText] = useState(''); const [arming, setArming] = useState(false); const [error, setError] = useState(); + const [reconciliation, setReconciliation] = + useState(); + const armGenerationRef = useRef(0); // Each opening starts from empty: the previous Session's condition is not a // useful default for this one, and a stale error would outlive its cause. - useEffect(() => { + useLayoutEffect(() => { + armGenerationRef.current += 1; if (props.sessionId === undefined) return; setCondition(''); setMaxIterationsText(''); setTokenBudgetText(''); setError(undefined); setArming(false); + setReconciliation(undefined); }, [props.sessionId]); const sessionId = props.sessionId; const maxIterations = readGoalBudget(maxIterationsText, 1, GOAL_MAX_ITERATIONS_LIMIT); const tokenBudget = readGoalBudget(tokenBudgetText, GOAL_TOKEN_BUDGET_MINIMUM); + const locked = reconciliation !== undefined; const canSubmit = condition.trim().length > 0 && maxIterations.kind !== 'invalid' && tokenBudget.kind !== 'invalid' && - !arming; + !arming && + !locked; + + const reconciliationMessage = (() => { + if (!reconciliation) return undefined; + switch (reconciliation.kind) { + case 'matching_goal': + return copy.reconciledMatching( + reconciliation.goal.condition, + copy.statusLabels[reconciliation.goal.status], + ); + case 'different_goal': + return copy.reconciledDifferent( + reconciliation.goal.condition, + copy.statusLabels[reconciliation.goal.status], + ); + case 'no_goal': + return copy.reconciledNoGoal; + case 'unavailable': + return copy.reconciliationUnavailable; + } + })(); async function arm(): Promise { if (!sessionId || !canSubmit) return; + const armGeneration = ++armGenerationRef.current; setArming(true); setError(undefined); try { - await window.maka.goal.arm(sessionId, { + const outcome = await window.maka.goal.arm(sessionId, { condition: condition.trim(), maxIterations: maxIterations.kind === 'value' ? maxIterations.value : null, tokenBudget: tokenBudget.kind === 'value' ? tokenBudget.value : null, }); - props.onClose(); + if (armGenerationRef.current !== armGeneration) return; + const action = interpretGoalArmOutcome(outcome); + if (action.action === 'close') { + props.onClose(); + } else { + setReconciliation(action.notice); + } } catch (cause) { + if (armGenerationRef.current !== armGeneration) return; setError(localizedShellErrorMessage(cause, copy.failedFallback, locale)); } finally { - setArming(false); + if (armGenerationRef.current === armGeneration) setArming(false); } } @@ -122,6 +161,9 @@ export function GoalDialog(props: { {copy.description} + {reconciliationMessage ? ( + {reconciliationMessage} + ) : null}