{"version":3,"file":"wait-subscriptions.d.ts","sourceRoot":"","sources":["../../../../src/runs/background/wait-subscriptions.ts"],"names":[],"mappings":"AAGA,OAAO,KAAK,EAAE,YAAY,EAAE,MAAM,2BAA2B,CAAC;AAE9D,OAAO,EAQN,KAAK,aAAa,EAClB,KAAK,sBAAsB,EAC3B,MAAM,uBAAuB,CAAC;AAM/B,MAAM,WAAW,wBAAwB;IACxC,UAAU,EAAE,OAAO,GAAG,YAAY,CAAC;IACnC,KAAK,EAAE,MAAM,CAAC;IACd,WAAW,EAAE,MAAM,CAAC;IACpB,SAAS,EAAE,MAAM,CAAC;CAClB;AAED,MAAM,WAAW,uBAAuB;IACvC,GAAG,CAAC,KAAK,EAAE,wBAAwB,GAAG,sBAAsB,CAAC;IAC7D,OAAO,IAAI,IAAI,CAAC;IAChB,SAAS,IAAI,IAAI,CAAC;IAClB,OAAO,IAAI,IAAI,CAAC;CAChB;AAED,UAAU,8BAA8B;IACvC,YAAY,CAAC,EAAE,MAAM,CAAC;IACtB,UAAU,CAAC,EAAE,MAAM,CAAC;IACpB,gBAAgB,CAAC,EAAE,MAAM,CAAC;IAC1B,GAAG,CAAC,EAAE,MAAM,MAAM,CAAC;IACnB,cAAc,CAAC,EAAE,MAAM,CAAC;IACxB,IAAI,CAAC,EAAE,CAAC,GAAG,EAAE,MAAM,EAAE,MAAM,CAAC,EAAE,MAAM,CAAC,OAAO,GAAG,CAAC,KAAK,OAAO,CAAC;CAC7D;AAqCD,wBAAgB,uBAAuB,CACtC,KAAK,EAAE,IAAI,CAAC,aAAa,EAAE,mBAAmB,CAAC,EAC/C,GAAG,SAAa,GACd,MAAM,GAAG,SAAS,CAYpB;AAED,wBAAgB,6BAA6B,CAC5C,EAAE,EAAE,IAAI,CAAC,YAAY,EAAE,QAAQ,GAAG,aAAa,CAAC,EAChD,KAAK,EAAE,aAAa,EACpB,OAAO,GAAE,8BAAmC,GAC1C,uBAAuB,CAmMzB","sourcesContent":["import { randomUUID } from \"node:crypto\";\nimport * as fs from \"node:fs\";\nimport * as path from \"node:path\";\nimport type { ExtensionAPI } from \"@lpb-work/pi-coding-agent\";\nimport { writeAtomicJson } from \"../../shared/atomic-json.ts\";\nimport {\n\tDIRS,\n\tINTERCOM_DETACH_REQUEST_EVENT,\n\tSUBAGENT_ASYNC_COMPLETE_EVENT,\n\tSUBAGENT_CONTROL_EVENT,\n\tSUBAGENT_CONTROL_INTERCOM_EVENT,\n\tSUBAGENT_FOREGROUND_COMPLETE_EVENT,\n\tSUBAGENT_RESULT_INTERCOM_EVENT,\n\ttype SubagentState,\n\ttype WaitSubscriptionRecord,\n} from \"../../shared/types.ts\";\nimport { listAsyncRuns } from \"./async-status.ts\";\n\nconst SUBSCRIPTION_VERSION = 1;\nconst RECONCILE_INTERVAL_MS = 1_000;\n\nexport interface ArmWaitSubscriptionInput {\n\ttargetKind: \"async\" | \"foreground\";\n\trunId: string;\n\trequestedId: string;\n\ttimeoutMs: number;\n}\n\nexport interface WaitSubscriptionManager {\n\tarm(input: ArmWaitSubscriptionInput): WaitSubscriptionRecord;\n\trestore(): void;\n\treconcile(): void;\n\tdispose(): void;\n}\n\ninterface WaitSubscriptionManagerOptions {\n\tasyncDirRoot?: string;\n\tresultsDir?: string;\n\tsubscriptionsDir?: string;\n\tnow?: () => number;\n\tpollIntervalMs?: number;\n\tkill?: (pid: number, signal?: NodeJS.Signals | 0) => boolean;\n}\n\nfunction isNotFound(error: unknown): boolean {\n\treturn (\n\t\ttypeof error === \"object\" &&\n\t\terror !== null &&\n\t\t\"code\" in error &&\n\t\t(error as NodeJS.ErrnoException).code === \"ENOENT\"\n\t);\n}\n\nfunction parseRecord(value: unknown): WaitSubscriptionRecord | undefined {\n\tif (!value || typeof value !== \"object\" || Array.isArray(value)) return undefined;\n\tconst record = value as Partial<WaitSubscriptionRecord>;\n\tif (\n\t\trecord.version !== SUBSCRIPTION_VERSION ||\n\t\ttypeof record.token !== \"string\" ||\n\t\t!/^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(record.token) ||\n\t\ttypeof record.sessionId !== \"string\" ||\n\t\t(record.targetKind !== \"async\" && record.targetKind !== \"foreground\") ||\n\t\ttypeof record.runId !== \"string\" ||\n\t\ttypeof record.requestedId !== \"string\" ||\n\t\ttypeof record.createdAt !== \"number\" ||\n\t\ttypeof record.expiresAt !== \"number\"\n\t)\n\t\treturn undefined;\n\treturn record as WaitSubscriptionRecord;\n}\n\nfunction needsAttention(run: ReturnType<typeof listAsyncRuns>[number]): boolean {\n\treturn run.activityState === \"needs_attention\" || run.steps.some((step) => step.activityState === \"needs_attention\");\n}\n\nfunction subscriptionFile(dir: string, token: string): string {\n\treturn path.join(dir, `${token}.json`);\n}\n\nexport function formatWaitSubscriptions(\n\tstate: Pick<SubagentState, \"waitSubscriptions\">,\n\tnow = Date.now(),\n): string | undefined {\n\tconst subscriptions = [...(state.waitSubscriptions?.values() ?? [])].sort(\n\t\t(left, right) => left.createdAt - right.createdAt,\n\t);\n\tif (subscriptions.length === 0) return undefined;\n\tconst lines = [`Armed wait subscriptions (${subscriptions.length}):`];\n\tfor (const record of subscriptions) {\n\t\tlines.push(\n\t\t\t`- ${record.token}: ${record.targetKind} run ${record.runId}, timeout in ${Math.max(0, record.expiresAt - now)}ms`,\n\t\t);\n\t}\n\treturn lines.join(\"\\n\");\n}\n\nexport function createWaitSubscriptionManager(\n\tpi: Pick<ExtensionAPI, \"events\" | \"sendMessage\">,\n\tstate: SubagentState,\n\toptions: WaitSubscriptionManagerOptions = {},\n): WaitSubscriptionManager {\n\tconst asyncDirRoot = options.asyncDirRoot ?? DIRS.async;\n\tconst resultsDir = options.resultsDir ?? DIRS.results;\n\tconst subscriptionsDir = options.subscriptionsDir ?? path.join(path.dirname(asyncDirRoot), \"wait-subscriptions\");\n\tconst now = options.now ?? Date.now;\n\tconst subscriptions = state.waitSubscriptions ?? new Map<string, WaitSubscriptionRecord>();\n\tstate.waitSubscriptions = subscriptions;\n\tconst unresolvedRestoredForegroundTokens = new Set<string>();\n\tlet disposed = false;\n\n\tconst remove = (record: WaitSubscriptionRecord) => {\n\t\ttry {\n\t\t\tfs.unlinkSync(subscriptionFile(subscriptionsDir, record.token));\n\t\t} catch (error) {\n\t\t\tif (!isNotFound(error)) throw error;\n\t\t}\n\t\tsubscriptions.delete(record.token);\n\t\tunresolvedRestoredForegroundTokens.delete(record.token);\n\t};\n\n\tconst settle = (record: WaitSubscriptionRecord, outcome: string, detail: string) => {\n\t\tif (disposed || state.currentSessionId !== record.sessionId) return;\n\t\ttry {\n\t\t\tremove(record);\n\t\t} catch (error) {\n\t\t\tconsole.error(`Failed to clear wait subscription '${record.token}'; it remains armed:`, error);\n\t\t\treturn;\n\t\t}\n\t\ttry {\n\t\t\tpi.sendMessage(\n\t\t\t\t{\n\t\t\t\t\tcustomType: \"subagent-wait-subscription\",\n\t\t\t\t\tcontent: `Wait subscription ${record.token} fired for run ${record.runId}: ${outcome}. ${detail}`,\n\t\t\t\t\tdisplay: true,\n\t\t\t\t\tdetails: { token: record.token, runId: record.runId, outcome },\n\t\t\t\t},\n\t\t\t\t{ triggerTurn: true },\n\t\t\t);\n\t\t} catch (error) {\n\t\t\tconsole.error(`Failed to deliver wait subscription '${record.token}' after clearing it:`, error);\n\t\t}\n\t};\n\n\tconst reconcileRecord = (record: WaitSubscriptionRecord) => {\n\t\tif (record.sessionId !== state.currentSessionId) return;\n\t\tif (now() >= record.expiresAt) {\n\t\t\tsettle(\n\t\t\t\trecord,\n\t\t\t\t\"timed out\",\n\t\t\t\t\"The targeted run may still be active; inspect its status before taking follow-up action.\",\n\t\t\t);\n\t\t\treturn;\n\t\t}\n\t\tif (record.targetKind === \"foreground\") {\n\t\t\tconst run = state.foregroundRuns?.get(record.runId);\n\t\t\tif (!run && unresolvedRestoredForegroundTokens.has(record.token)) return;\n\t\t\tif (run) unresolvedRestoredForegroundTokens.delete(record.token);\n\t\t\tif (!run || run.sessionId !== record.sessionId) {\n\t\t\t\tsettle(\n\t\t\t\t\trecord,\n\t\t\t\t\t\"could not be reconciled\",\n\t\t\t\t\t\"The remembered foreground run disappeared before completion was confirmed.\",\n\t\t\t\t);\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tconst detached = run.children.filter((child) => child.status === \"detached\");\n\t\t\tif (detached.some((child) => child.activityState === \"needs_attention\")) {\n\t\t\t\tsettle(record, \"needs attention\", \"Reply to the pending supervisor request or inspect the run status.\");\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tif (detached.length === 0) {\n\t\t\t\tconst failed = run.children.some((child) => child.status === \"failed\");\n\t\t\t\tsettle(record, failed ? \"failed\" : \"completed\", \"Inspect the run status for its final output.\");\n\t\t\t}\n\t\t\treturn;\n\t\t}\n\n\t\tconst runs = listAsyncRuns(asyncDirRoot, {\n\t\t\tsessionId: record.sessionId,\n\t\t\trunId: record.runId,\n\t\t\tresultsDir,\n\t\t\tkill: options.kill,\n\t\t\tnow,\n\t\t});\n\t\tconst run = runs.find((candidate) => candidate.id === record.runId);\n\t\tif (!run) {\n\t\t\tsettle(\n\t\t\t\trecord,\n\t\t\t\t\"could not be reconciled\",\n\t\t\t\t\"The exact async run disappeared before a terminal result was confirmed.\",\n\t\t\t);\n\t\t\treturn;\n\t\t}\n\t\tif (needsAttention(run)) {\n\t\t\tsettle(record, \"needs attention\", \"Inspect the run status and answer any pending supervisor request.\");\n\t\t\treturn;\n\t\t}\n\t\tif (run.state !== \"queued\" && run.state !== \"running\") {\n\t\t\tsettle(\n\t\t\t\trecord,\n\t\t\t\trun.state === \"complete\" ? \"completed\" : run.state,\n\t\t\t\t\"Inspect the run status for its final output.\",\n\t\t\t);\n\t\t}\n\t};\n\n\tconst reconcile = () => {\n\t\tif (disposed || !state.currentSessionId) return;\n\t\tfor (const record of [...subscriptions.values()]) {\n\t\t\ttry {\n\t\t\t\treconcileRecord(record);\n\t\t\t} catch (error) {\n\t\t\t\tconsole.error(`Failed to reconcile wait subscription '${record.token}':`, error);\n\t\t\t\tsettle(\n\t\t\t\t\trecord,\n\t\t\t\t\t\"reconciliation failed\",\n\t\t\t\t\t\"The targeted run could not be reconciled. Inspect its status before taking follow-up action.\",\n\t\t\t\t);\n\t\t\t}\n\t\t}\n\t};\n\n\tconst wakeChannels = [\n\t\tINTERCOM_DETACH_REQUEST_EVENT,\n\t\tSUBAGENT_ASYNC_COMPLETE_EVENT,\n\t\tSUBAGENT_FOREGROUND_COMPLETE_EVENT,\n\t\tSUBAGENT_CONTROL_EVENT,\n\t\tSUBAGENT_CONTROL_INTERCOM_EVENT,\n\t\tSUBAGENT_RESULT_INTERCOM_EVENT,\n\t];\n\tconst unsubscribes = wakeChannels.map((channel) => pi.events.on(channel, reconcile));\n\tconst interval = setInterval(reconcile, options.pollIntervalMs ?? RECONCILE_INTERVAL_MS);\n\tinterval.unref?.();\n\n\treturn {\n\t\tarm(input) {\n\t\t\tconst sessionId = state.currentSessionId;\n\t\t\tif (!sessionId) throw new Error(\"A wait subscription requires an active session identity.\");\n\t\t\tconst createdAt = now();\n\t\t\tconst record: WaitSubscriptionRecord = {\n\t\t\t\tversion: SUBSCRIPTION_VERSION,\n\t\t\t\ttoken: randomUUID(),\n\t\t\t\tsessionId,\n\t\t\t\ttargetKind: input.targetKind,\n\t\t\t\trunId: input.runId,\n\t\t\t\trequestedId: input.requestedId,\n\t\t\t\tcreatedAt,\n\t\t\t\texpiresAt: createdAt + input.timeoutMs,\n\t\t\t};\n\t\t\tfs.mkdirSync(subscriptionsDir, { recursive: true });\n\t\t\twriteAtomicJson(subscriptionFile(subscriptionsDir, record.token), record);\n\t\t\tsubscriptions.set(record.token, record);\n\t\t\treturn record;\n\t\t},\n\t\trestore() {\n\t\t\tsubscriptions.clear();\n\t\t\tunresolvedRestoredForegroundTokens.clear();\n\t\t\tif (!state.currentSessionId) return;\n\t\t\tlet files: string[];\n\t\t\ttry {\n\t\t\t\tfiles = fs.readdirSync(subscriptionsDir).filter((file) => file.endsWith(\".json\"));\n\t\t\t} catch (error) {\n\t\t\t\tif (isNotFound(error)) return;\n\t\t\t\tthrow error;\n\t\t\t}\n\t\t\tfor (const file of files) {\n\t\t\t\ttry {\n\t\t\t\t\tconst record = parseRecord(JSON.parse(fs.readFileSync(path.join(subscriptionsDir, file), \"utf-8\")));\n\t\t\t\t\tif (!record || file !== `${record.token}.json`) continue;\n\t\t\t\t\tif (record.sessionId === state.currentSessionId) {\n\t\t\t\t\t\tsubscriptions.set(record.token, record);\n\t\t\t\t\t\tif (record.targetKind === \"foreground\" && !state.foregroundRuns?.has(record.runId))\n\t\t\t\t\t\t\tunresolvedRestoredForegroundTokens.add(record.token);\n\t\t\t\t\t}\n\t\t\t\t} catch (error) {\n\t\t\t\t\tconsole.error(`Ignoring invalid wait subscription '${path.join(subscriptionsDir, file)}':`, error);\n\t\t\t\t}\n\t\t\t}\n\t\t\treconcile();\n\t\t},\n\t\treconcile,\n\t\tdispose() {\n\t\t\tif (disposed) return;\n\t\t\tdisposed = true;\n\t\t\tclearInterval(interval);\n\t\t\tfor (const unsubscribe of unsubscribes) {\n\t\t\t\ttry {\n\t\t\t\t\tunsubscribe();\n\t\t\t\t} catch {\n\t\t\t\t\t/* best effort */\n\t\t\t\t}\n\t\t\t}\n\t\t\tsubscriptions.clear();\n\t\t},\n\t};\n}\n"]}