diff --git a/src/chat-swarm-runtime.test.ts b/src/chat-swarm-runtime.test.ts index 113c08edc..7a06ddc91 100644 --- a/src/chat-swarm-runtime.test.ts +++ b/src/chat-swarm-runtime.test.ts @@ -139,7 +139,10 @@ class FakeManagedAdapter implements ChatSwarmManagedCarrierAdapter { } } -function fixture(workerLimit = 5) { +function fixture( + workerLimit = 5, + transport: "cdp" | "opencli" = "cdp", +) { const root = mkdtempSync(join(tmpdir(), "devspace-runtime-117-")); const store = new ChatSwarmStore(root); const coordinator = new ChatSwarmCoordinator(store); @@ -150,6 +153,7 @@ function fixture(workerLimit = 5) { const env: NodeJS.ProcessEnv = { DEVSPACE_CHAT_SWARM_RUNTIME: "1", DEVSPACE_CHAT_SWARM_PROJECT_URL: "https://chatgpt.com/g/g-p-runtime-test/project", + DEVSPACE_CHAT_SWARM_TRANSPORT: transport, DEVSPACE_CHAT_SWARM_POOL_DEFAULT: "3", DEVSPACE_CHAT_SWARM_RUNTIME_TIMEOUT_MS: "5000", DEVSPACE_CHAT_SWARM_BOOTSTRAP_WAIT_MS: "5000", @@ -252,7 +256,7 @@ test("runtime config selects OpenCLI explicitly while preserving CDP as the defa assert.equal(fallback.transport, "cdp"); }); -test("OpenCLI provisioning captures transport and authenticated peer identities separately", async () => { +test("OpenCLI provisioning durably exposes the conversation handle before peer identity readback", async () => { const driver = openCliDriverForTest(); const calls: string[][] = []; const peerFingerprint = "a".repeat(64); @@ -275,15 +279,23 @@ test("OpenCLI provisioning captures transport and authenticated peer identities response: "", }]; }; - const evidence = await driver.createManagedConversation( + const observed = await driver.createManagedConversation( "https://chatgpt.com/g/g-p-runtime-test/project", new Date(Date.now() + 1_000).toISOString(), ); assert.equal( - evidence.conversationUrl, + observed.conversationUrl, "https://chatgpt.com/g/g-p-runtime-test/c/opencli-managed-01", ); - assert.equal(evidence.conversationFingerprint, fingerprint("opencli-managed-01")); + assert.equal(observed.conversationFingerprint, fingerprint("opencli-managed-01")); + assert.equal(observed.authenticatedPeerFingerprint, undefined); + assert.equal(observed.appBinding, "UNKNOWN"); + assert.equal(calls.filter((call) => call[1] === "detail").length, 0); + + const evidence = await driver.resolveManagedConversationIdentity!( + observed.conversationUrl, + new Date(Date.now() + 1_000).toISOString(), + ); assert.equal(evidence.authenticatedPeerFingerprint, peerFingerprint); assert.notEqual(evidence.conversationFingerprint, peerFingerprint); assert.equal(evidence.appBinding, "READY"); @@ -297,7 +309,7 @@ test("OpenCLI provisioning captures transport and authenticated peer identities assert.equal(calls[0]?.[calls[0]!.indexOf("--site-session") + 1], "ephemeral"); }); -test("OpenCLI provisioning fails closed when the authenticated peer probe is malformed", async () => { +test("OpenCLI peer identity resolution fails closed while preserving the observed conversation handle", async () => { const driver = openCliDriverForTest(); let probePrompt = ""; (driver as any).runJson = async (args: string[]) => { @@ -314,9 +326,14 @@ test("OpenCLI provisioning fails closed when the authenticated peer probe is mal response: "", }]; }; + const observed = await driver.createManagedConversation( + "https://chatgpt.com/g/g-p-runtime-test/project", + new Date(Date.now() + 1_000).toISOString(), + ); + assert.equal(observed.conversationFingerprint, fingerprint("opencli-managed-bad-peer")); await assert.rejects( - () => driver.createManagedConversation( - "https://chatgpt.com/g/g-p-runtime-test/project", + () => driver.resolveManagedConversationIdentity!( + observed.conversationUrl, new Date(Date.now() + 1_000).toISOString(), ), /peer identity probe did not return a valid fingerprint/, @@ -404,6 +421,31 @@ test("runtime provision lease spans all bounded provisioning phases", async () = } }); +test("OpenCLI runtime lease spans handle, peer identity, bootstrap, and bound-wait phases", async () => { + const f = fixture(5, "opencli"); + try { + let observedLeaseMs = 0; + f.adapter.onBootstrap = (operationId, rawIdentity) => { + const operation = f.registry.getProvision(operationId); + assert.ok(operation); + observedLeaseMs = + Date.parse(operation.request.expiresAt) - + Date.parse(operation.request.requestedAt); + f.manager.bootstrap({ "openai/session": rawIdentity }, operationId); + }; + + const status = await f.manager.ensure(f.owner, f.swarm.id, 1); + assert.equal( + observedLeaseMs, + f.manager.runtimeConfig.operationTimeoutMs * 3 + + f.manager.runtimeConfig.bootstrapWaitMs, + ); + assert.equal(status.slots[0]?.state, "PARKED"); + } finally { + cleanup(f); + } +}); + test("concurrent runtime ensure creates only missing managed workers and exact replay creates no duplicates", async () => { const f = fixture(); try { @@ -616,6 +658,125 @@ test("response loss stops the current ensure at the first uncertain carrier and } }); +test("deferred peer identity resolution binds bootstrap authority to the authenticated peer", async () => { + const f = fixture(); + const peerRawIdentity = "authenticated-managed-peer-1"; + const peerFingerprint = fingerprint(peerRawIdentity); + (f.adapter as any).resolveProvisionIdentity = async (input: { + conversationUrl: string; + }): Promise => ({ + conversationUrl: input.conversationUrl, + conversationFingerprint: fingerprint("managed-conversation-1"), + authenticatedPeerFingerprint: peerFingerprint, + appBinding: "READY", + }); + f.adapter.onBootstrap = (operationId) => { + f.manager.bootstrap({ "openai/session": peerRawIdentity }, operationId); + }; + try { + const status = await f.manager.ensure(f.owner, f.swarm.id, 1); + assert.equal(status.slots[0]?.state, "PARKED"); + assert.equal( + status.slots[0]?.conversationFingerprint, + fingerprint("managed-conversation-1"), + ); + assert.equal(status.slots[0]?.authenticatedPeerFingerprint, peerFingerprint); + assert.notEqual( + status.slots[0]?.conversationFingerprint, + status.slots[0]?.authenticatedPeerFingerprint, + ); + assert.equal(f.adapter.provisionCalls, 1); + assert.equal(f.adapter.bootstrapCalls, 1); + } finally { + cleanup(f); + } +}); + +test("ensure resumes peer identity readback from a durably observed handle without reprovisioning", async () => { + const f = fixture(); + const slot = f.registry.ensureSlot( + f.swarm.id, + 1, + "https://chatgpt.com/g/g-p-runtime-test/project", + fingerprint(f.manager.runtimeConfig.browserProfileDir), + ); + const prepared = f.registry.prepareProvision(slot, 30_000); + assert.ok(prepared.operation); + assert.equal(f.registry.claimProvision(prepared.operation!.operationId), true); + const transportRawIdentity = "managed-conversation-1"; + const transportFingerprint = fingerprint(transportRawIdentity); + f.registry.markCarrierObserved(prepared.operation!.operationId, { + conversationUrl: `https://chatgpt.com/c/${transportRawIdentity}`, + conversationFingerprint: transportFingerprint, + appBinding: "UNKNOWN", + }); + + const peerRawIdentity = "resumed-authenticated-peer-1"; + const peerFingerprint = fingerprint(peerRawIdentity); + let resolveCalls = 0; + (f.adapter as any).resolveProvisionIdentity = async (input: { + conversationUrl: string; + }): Promise => { + resolveCalls += 1; + assert.equal(input.conversationUrl, `https://chatgpt.com/c/${transportRawIdentity}`); + return { + conversationUrl: input.conversationUrl, + conversationFingerprint: transportFingerprint, + authenticatedPeerFingerprint: peerFingerprint, + appBinding: "READY", + }; + }; + f.adapter.onBootstrap = (operationId) => { + f.manager.bootstrap({ "openai/session": peerRawIdentity }, operationId); + }; + + try { + const status = await f.manager.ensure(f.owner, f.swarm.id, 1); + assert.equal(status.slots[0]?.state, "PARKED"); + assert.equal(status.slots[0]?.authenticatedPeerFingerprint, peerFingerprint); + assert.equal(f.adapter.provisionCalls, 0); + assert.equal(resolveCalls, 1); + assert.equal(f.adapter.bootstrapCalls, 1); + } finally { + cleanup(f); + } +}); + +test("peer identity timeout preserves the exact observed conversation handle and stops further provisioning", async () => { + const f = fixture(); + (f.adapter as any).resolveProvisionIdentity = async (input: { + conversationUrl: string; + }) => { + assert.match(input.conversationUrl, /managed-conversation-1$/); + throw new Error("peer identity readback timed out"); + }; + try { + const first = await f.manager.ensure(f.owner, f.swarm.id, 3); + assert.equal(first.state, "RECONCILE_REQUIRED"); + assert.equal(first.slots.length, 1); + assert.equal(first.slots[0]!.state, "RECONCILE_REQUIRED"); + assert.match(first.slots[0]!.conversationUrl ?? "", /managed-conversation-1$/); + assert.equal( + first.slots[0]!.conversationFingerprint, + fingerprint("managed-conversation-1"), + ); + assert.equal(first.slots[0]!.authenticatedPeerFingerprint, undefined); + assert.equal(first.slots[0]!.blocker, "peer identity readback timed out"); + assert.equal(f.adapter.provisionCalls, 1); + assert.equal(f.adapter.bootstrapCalls, 0); + + const operation = f.registry.getProvision(first.slots[0]!.lastOperationId!); + assert.equal(operation?.status, "outcome_unknown"); + assert.match(operation?.receipt?.conversationUrl ?? "", /managed-conversation-1$/); + assert.equal( + operation?.receipt?.conversationFingerprint, + fingerprint("managed-conversation-1"), + ); + } finally { + cleanup(f); + } +}); + test("managed registry survives reopen without duplicating logical workers", async () => { const f = fixture(); try { diff --git a/src/chat-swarm-runtime.ts b/src/chat-swarm-runtime.ts index c2a820c92..df99ea54e 100644 --- a/src/chat-swarm-runtime.ts +++ b/src/chat-swarm-runtime.ts @@ -128,6 +128,7 @@ interface ProvisionReceipt { schema: typeof PROVISION_RECEIPT_SCHEMA; disposition: | "PREPARED" + | "CARRIER_OBSERVED" | "CARRIER_CREATED" | "BOOTSTRAPPING" | "BOUND" @@ -214,6 +215,10 @@ export interface ChatSwarmManagedCarrierAdapter extends ChatSwarmCarrierAdapter projectUrl: string; deadlineAt: string; }): Promise; + resolveProvisionIdentity?(input: { + conversationUrl: string; + deadlineAt: string; + }): Promise; bootstrap(input: { operationId: string; swarmId: string; @@ -241,6 +246,10 @@ export interface MacWebDriver { projectUrl: string, deadlineAt: string, ): Promise; + resolveManagedConversationIdentity?( + conversationUrl: string, + deadlineAt: string, + ): Promise; sendPrompt( conversationUrl: string, prompt: string, @@ -353,7 +362,8 @@ function profileId(path: string): string { return createHash("sha256").update(resolve(path)).digest("hex"); } function runtimeProvisionLeaseMs(config: ChatSwarmRuntimeConfig): number { - return config.operationTimeoutMs * 2 + config.bootstrapWaitMs; + const boundedOperationPhases = config.transport === "opencli" ? 3 : 2; + return config.operationTimeoutMs * boundedOperationPhases + config.bootstrapWaitMs; } function slotAttemptKey(swarmId: string, runtimeSlot: number): string { return `chat-swarm-runtime-slot:${swarmId}:${runtimeSlot}`; @@ -646,6 +656,62 @@ export class ChatSwarmRuntimeStore { return result.changes === 1; } + markCarrierObserved( + operationId: string, + evidence: ManagedConversationEvidence, + ): ManagedCarrierSlot { + assertFingerprint(evidence.conversationFingerprint, "conversation fingerprint"); + const tx = this.database.sqlite.transaction(() => { + const operation = this.requireProvision(operationId); + if (operation.status === "outcome_unknown") { + throw new ChatSwarmError( + "RECONCILIATION_REQUIRED", + "provision outcome is unknown; do not create another carrier", + ); + } + if (operation.status !== "started") { + throw new ChatSwarmError("INVALID_STATE", "provision operation is not active"); + } + const conflicting = this.getSlotByFingerprint(evidence.conversationFingerprint); + if ( + conflicting && + (conflicting.swarmId !== operation.request.swarmId || + conflicting.runtimeSlot !== operation.request.runtimeSlot) + ) { + throw new ChatSwarmError( + "OWNERSHIP_CONFLICT", + "conversation is already managed by another runtime slot", + ); + } + const observedAt = nowIso(); + const receipt: ProvisionReceipt = { + schema: PROVISION_RECEIPT_SCHEMA, + disposition: "CARRIER_OBSERVED", + conversationUrl: evidence.conversationUrl, + conversationFingerprint: evidence.conversationFingerprint, + remoteMayContinue: true, + observedAt, + }; + this.updateProvision(operationId, "started", receipt); + const slot = this.getSlot(operation.request.swarmId, operation.request.runtimeSlot)!; + this.updateSlotReceipt( + slot, + { + schema: SLOT_RECEIPT_SCHEMA, + generation: operation.request.generation, + state: "PROVISIONING", + conversationUrl: evidence.conversationUrl, + conversationFingerprint: evidence.conversationFingerprint, + lastOperationId: operationId, + updatedAt: observedAt, + }, + "started", + ); + return this.getSlot(operation.request.swarmId, operation.request.runtimeSlot)!; + }); + return tx.immediate(); + } + markCarrierCreated( operationId: string, evidence: ManagedConversationEvidence, @@ -1435,6 +1501,18 @@ export class OpenCliMacWebDriver implements MacWebDriver { if (!conversationUrl) { throw new Error("OpenCLI did not return a ChatGPT conversation URL"); } + return { + conversationUrl, + conversationFingerprint: conversationFingerprintFromUrl(conversationUrl), + appBinding: "UNKNOWN", + }; + } + + async resolveManagedConversationIdentity( + conversationUrl: string, + deadlineAt: string, + ): Promise { + const probePrompt = this.peerIdentityProbePrompt(); await this.waitForConversationIdle(conversationUrl, deadlineAt); const authenticatedPeerFingerprint = this.peerFingerprintFromDetail( await this.detail(conversationUrl, deadlineAt), @@ -1998,6 +2076,10 @@ export function conversationFingerprintFromUrl(value: string): string { export class MacWebChatCarrierAdapter implements ChatSwarmManagedCarrierAdapter { readonly kind = "mac_web_chatgpt"; readonly configHash: string; + readonly resolveProvisionIdentity?: (input: { + conversationUrl: string; + deadlineAt: string; + }) => Promise; constructor( readonly config: ChatSwarmRuntimeConfig, @@ -2015,6 +2097,13 @@ export class MacWebChatCarrierAdapter implements ChatSwarmManagedCarrierAdapter appLabel: config.appLabel, kind: this.kind, }); + if (this.driver.resolveManagedConversationIdentity) { + this.resolveProvisionIdentity = (input) => + this.driver.resolveManagedConversationIdentity!( + input.conversationUrl, + input.deadlineAt, + ); + } } capabilities() { @@ -2299,7 +2388,11 @@ export class ChatSwarmRuntimeManager { continue; } operation = this.registry.getProvision(operation.operationId)!; - } else if (operation.status === "started" && !prepared.created) { + } else if ( + operation.status === "started" && + !prepared.created && + !operation.receipt?.conversationUrl + ) { await this.waitForPeerInvocation(slot).catch(() => undefined); continue; } @@ -2323,7 +2416,38 @@ export class ChatSwarmRuntimeManager { ); break; } - slot = this.registry.markCarrierCreated(operation.operationId, evidence); + if (this.adapter.resolveProvisionIdentity) { + slot = this.registry.markCarrierObserved(operation.operationId, evidence); + } else { + slot = this.registry.markCarrierCreated(operation.operationId, evidence); + } + operation = this.registry.getProvision(operation.operationId)!; + } + + if ( + operation.status === "started" && + operation.receipt?.conversationUrl && + this.adapter.resolveProvisionIdentity + ) { + let resolvedEvidence: ManagedConversationEvidence; + try { + resolvedEvidence = await this.adapter.resolveProvisionIdentity({ + conversationUrl: operation.receipt.conversationUrl, + deadlineAt: new Date( + Date.now() + this.runtimeConfig.operationTimeoutMs, + ).toISOString(), + }); + } catch (error) { + this.registry.markProvisionUnknown( + operation.operationId, + error instanceof Error ? error.message : String(error), + ); + break; + } + slot = this.registry.markCarrierCreated( + operation.operationId, + resolvedEvidence, + ); operation = this.registry.getProvision(operation.operationId)!; }