Merge v3-t54: watchdog as a world extension (T54, P3-l)
# Conflicts: # host/js/package.json
This commit is contained in:
@@ -0,0 +1,19 @@
|
||||
# Watchdog
|
||||
|
||||
Watchdog observes failed model generations. It ignores user stops and reply-expected board inputs, which messages handles itself.
|
||||
|
||||
A temporary provider failure creates a durable `watchdog.failure` task. The task waits 60 seconds before one continuation.
|
||||
A failed continuation does not receive another retry. Permanent errors report immediately.
|
||||
If another run remains active at the deadline, watchdog does not submit a continuation.
|
||||
|
||||
Reports use the messages service with a stable request ID. Owned children report to their owner. Other conversations report to @ultimate.
|
||||
A failure of @ultimate itself produces a notice instead. The service retains the latest 20 reported failures.
|
||||
|
||||
Receipts and a high-water mark prevent repeated recovery. Fresh installation excludes historical failures.
|
||||
On reopen, watchdog recovers failures that its observer missed. Retry inputs also use stable request IDs.
|
||||
|
||||
Before @ultimate receives an input, watchdog can add an idle-branch note. It does not modify the system prompt.
|
||||
The note excludes active branches, concluded branches, user questions, and queued landings when workspaces is available.
|
||||
It waits 30 minutes, or two hours while a branch awaits an agent reply. It repeats only when the branch IDs change.
|
||||
|
||||
The tests use the Harness clock and timer options. They advance the durable deadline without a 60-second wall-clock wait.
|
||||
@@ -0,0 +1,13 @@
|
||||
/** Watchdog: the failed runs it reported, newest first, for other extensions to summarize. */
|
||||
import { defineApi } from '../messages/compat.ts'
|
||||
import type { Failure } from './refs.ts'
|
||||
|
||||
/** A reported failure and when its run failed (epoch ms). */
|
||||
export type ReportedFailure = Failure & { at: number }
|
||||
|
||||
export interface WatchdogService {
|
||||
/** The last 20 reported failures whose run failed at or after `since` (epoch ms), newest first. */
|
||||
failures(since: number): Promise<ReportedFailure[]>
|
||||
}
|
||||
/** Reported failures; `want` it, since watchdog is optional. */
|
||||
export const WatchdogApi = defineApi<WatchdogService>()('watchdog')
|
||||
@@ -0,0 +1,102 @@
|
||||
/** Idle branch reminders, placed only before the primary agent input. */
|
||||
import { defineDoc, LiveDoc, ROOT_CONVERSATION_ID } from '@earendil-works/pi-durable'
|
||||
import type { ConversationId, Cursor, Tx } from '@earendil-works/pi-durable'
|
||||
import type { Host as ExtensionHost } from '../../host/js/extensions.ts'
|
||||
import { ConversationsApi } from '../conversations/api.ts'
|
||||
import { Branch } from '../lineage/api.ts'
|
||||
import { Messages } from '../messages/api.ts'
|
||||
|
||||
const Seen = defineDoc<{ key?: string }>({
|
||||
kind: 'watchdog.seen',
|
||||
version: 1,
|
||||
scope: 'conversation',
|
||||
history: 'latest',
|
||||
fork: 'initial',
|
||||
checkpointWhen: () => false,
|
||||
initial: () => ({}),
|
||||
})
|
||||
|
||||
export async function idleNote(
|
||||
tx: Tx,
|
||||
host: ExtensionHost,
|
||||
conversationId: ConversationId,
|
||||
now: number,
|
||||
) {
|
||||
const concluded = await host.need(ConversationsApi).concludedIn(tx)
|
||||
const workspaces = host.want({name: 'workspaces'})
|
||||
const queue = workspaces ? (await tx.doc(workspaces.doc, ROOT_CONVERSATION_ID)).queue : []
|
||||
const excluded = new Set([...concluded, ...queue.map((item) => item.conversationId)])
|
||||
const messages = (await tx.doc(Messages, ROOT_CONVERSATION_ID)).messages
|
||||
// A branch waiting on a reply gets 2 h; one that ended its run with nothing to wait on may have stalled: 30 min.
|
||||
const waiting = new Set<number>()
|
||||
for (const message of messages)
|
||||
if (
|
||||
typeof message.from === 'number' &&
|
||||
message.reply &&
|
||||
(message.state === 'sent' || message.state === 'delivered')
|
||||
)
|
||||
(message.to === 'user' ? excluded : waiting).add(message.from)
|
||||
const conversations = []
|
||||
let cursor: Cursor | undefined
|
||||
do {
|
||||
const page = await tx.scanConversations({}, 500, cursor)
|
||||
conversations.push(...page.items)
|
||||
cursor = page.next
|
||||
} while (cursor)
|
||||
// Owned descendants keep their branch working even when its coordinator is between turns.
|
||||
const children = new Map<ConversationId, ConversationId[]>()
|
||||
for (const conversation of conversations) {
|
||||
if (!conversation.owner) continue
|
||||
const owner = conversation.owner.conversationId
|
||||
const siblings = children.get(owner) ?? []
|
||||
siblings.push(conversation.id)
|
||||
children.set(owner, siblings)
|
||||
}
|
||||
const working = async (id: ConversationId): Promise<boolean> => {
|
||||
if ((await tx.doc(LiveDoc, id)).run) return true
|
||||
for (const child of children.get(id) ?? []) if (await working(child)) return true
|
||||
return false
|
||||
}
|
||||
const idle = []
|
||||
// Branches are ownerless, so subagents are skipped before any document read: this runs before every @ultimate input.
|
||||
for (const conversation of conversations) {
|
||||
if (conversation.owner || excluded.has(conversation.id)) continue
|
||||
if (!(await tx.doc(Branch, conversation.id)).branch) continue
|
||||
const metadata = await tx.doc(host.metadata, conversation.id)
|
||||
const age =
|
||||
now -
|
||||
Math.max(
|
||||
metadata.inputAt ?? metadata.answerAt ?? now,
|
||||
metadata.answerAt ?? metadata.inputAt ?? now,
|
||||
)
|
||||
const limit = waiting.has(conversation.id) ? 2 * 3_600_000 : 30 * 60_000
|
||||
if (age <= limit || (await working(conversation.id))) continue
|
||||
idle.push({
|
||||
id: conversation.id,
|
||||
title: metadata.title ?? 'Untitled',
|
||||
idle: age < 3_600_000 ? `${Math.floor(age / 60_000)}m` : `${Math.floor(age / 3_600_000)}h`,
|
||||
})
|
||||
}
|
||||
const key = idle.map((branch) => branch.id).join(',')
|
||||
const seen = await tx.doc(Seen, conversationId)
|
||||
if ((seen.key ?? '') === key) return undefined
|
||||
seen.key = key
|
||||
const content = idle.length
|
||||
? [
|
||||
'Open branches idle (30 min, or 2 h while awaiting a reply); check on each or conclude it:',
|
||||
...idle
|
||||
.slice(0, 10)
|
||||
.map((branch) => `- #${branch.id} ${branch.title} (idle ${branch.idle})`),
|
||||
...(idle.length > 10 ? [`and ${idle.length - 10} more`] : []),
|
||||
].join('\n')
|
||||
: 'No branches are idle.'
|
||||
return {
|
||||
entries: [
|
||||
{
|
||||
kind: 'watchdog.idle',
|
||||
data: { ids: idle.map((branch) => branch.id), text: content },
|
||||
model: [{ role: 'user' as const, content, timestamp: now }],
|
||||
},
|
||||
],
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
/** Watchdog failure cards: the failed conversation, error, last step and retry outcome. */
|
||||
export type Failure = {
|
||||
conversation: number
|
||||
title: string
|
||||
error: string
|
||||
step: string
|
||||
hint?: string
|
||||
retried: boolean
|
||||
}
|
||||
@@ -0,0 +1,137 @@
|
||||
/** pi-ai 1.5.0 retry classification and the commit observer adapter. */
|
||||
function buildProviderErrorPattern(patterns) {
|
||||
return new RegExp(patterns.join("|"), "i");
|
||||
}
|
||||
const NON_RETRYABLE_PROVIDER_LIMIT_ERROR_PATTERN = buildProviderErrorPattern([
|
||||
// OpenCode Go/free-tier limits returned as 429 JSON error types by OpenCode's
|
||||
// Zen API. These are subscription/account limits, not transient throttles.
|
||||
"GoUsageLimitError",
|
||||
"FreeUsageLimitError",
|
||||
// OpenCode Go subscription-limit text asks users to enable available-balance
|
||||
// usage after rolling/weekly/monthly limits are reached.
|
||||
"Monthly usage limit reached",
|
||||
"available balance",
|
||||
// Generic quota/budget/billing exhaustion. `insufficient_quota` is OpenAI's
|
||||
// quota/billing error code; the other strings cover common gateway wording.
|
||||
"insufficient_quota",
|
||||
"out of budget",
|
||||
"quota exceeded",
|
||||
"billing",
|
||||
// Sign in with ChatGPT: the subscription's shared usage limit, which resets
|
||||
// after hours rather than seconds.
|
||||
"subscription_sharing_usage_limit_exceeded",
|
||||
// Sign in with ChatGPT on a ChatGPT Business plan: a usage limit set by the
|
||||
// workspace admin. Only the admin can raise it.
|
||||
"subscription_sharing_workplace_individual_usage_limit_exceeded",
|
||||
// Zhipu/Z.ai (open.bigmodel.cn) error code 1308: the plan's usage limit is
|
||||
// reached, sent as HTTP 429. It resets after hours, not seconds (#11537).
|
||||
'"code"\\s*:\\s*"?1308(?![0-9])',
|
||||
]);
|
||||
const RETRYABLE_PROVIDER_ERROR_PATTERN = buildProviderErrorPattern([
|
||||
// Generic provider load, HTTP status, and server-side transient failures.
|
||||
"overloaded",
|
||||
"server_busy",
|
||||
"servers are currently busy",
|
||||
"currently experiencing high demand",
|
||||
"model is at capacity",
|
||||
"rate.?limit",
|
||||
"too many requests",
|
||||
// Status codes must not be part of a longer word or number, e.g. a request id or byte count (#11057).
|
||||
"(?<![a-z0-9])(?:429|500|502|503|504|520|524)(?![a-z0-9])",
|
||||
// Also matches "The service is temporarily unavailable" (#10965).
|
||||
"service.?(?:is.?)?(?:temporarily.?)?unavailable",
|
||||
"server.?error",
|
||||
"internal.?error",
|
||||
// Wrapper/provider text for transient upstream failures, including OpenRouter
|
||||
// "Provider returned error" responses (#2264) and gateway wording such as
|
||||
// "The upstream provider returned an error" (#10842).
|
||||
"provider.?returned.?(?:an.?)?error",
|
||||
"exceeded request buffer limit while retrying upstream",
|
||||
// Gateway failure relaying the upstream response stream, sent as an OpenAI
|
||||
// Responses `response.failed` event without an error code (#10920).
|
||||
"unable to forward upstream",
|
||||
// Other gateway upstream failures relayed as `response.failed` error codes (#10920).
|
||||
"upstream.?request.?failed",
|
||||
"upstream_stream_read_error",
|
||||
"stream was interrupted",
|
||||
"http/?2 stream failed",
|
||||
// Network, proxy, and fetch transport failures. This includes OpenAI Codex
|
||||
// raw-fetch failures such as "upstream connect", "connection refused", and
|
||||
// "reset before headers" (#733), plus OpenRouter connection drops (#3317).
|
||||
"network.?error",
|
||||
"connection.?error",
|
||||
"connection.?refused",
|
||||
"connection.?lost",
|
||||
"other side closed",
|
||||
"fetch failed",
|
||||
// node-fetch system errors, e.g. the Google ADC OAuth token refresh used by
|
||||
// google-vertex. The reason is often empty (#10996).
|
||||
"request to \\S+ failed, reason:",
|
||||
"getaddrinfo",
|
||||
"ENOTFOUND",
|
||||
"EAI_AGAIN",
|
||||
"upstream.?connect",
|
||||
"reset before headers",
|
||||
"socket hang up",
|
||||
"socket connection was closed",
|
||||
"timed? out",
|
||||
"timeout",
|
||||
"terminated",
|
||||
// WebSocket transports can report close/error text instead of HTTP/fetch text.
|
||||
"websocket.?closed",
|
||||
"websocket.?error",
|
||||
// Premature stream endings from SDKs and transports. Anthropic can throw
|
||||
// "stream ended without ..." and "Anthropic stream ended before message_stop"
|
||||
// (#4433); Bedrock/Smithy can throw an HTTP/2 no-response error (#3594).
|
||||
"ended without",
|
||||
"stream ended before message_stop",
|
||||
"stream ended before a terminal response event",
|
||||
"http2 request did not get a response",
|
||||
// Codex CLI mid-stream disconnect, relayed by proxies that wrap the Codex CLI,
|
||||
// e.g. "stream disconnected before completion: stream closed before response.completed" (#11057).
|
||||
"stream disconnected before completion",
|
||||
// Node ERR_HTTP2_STREAM_CANCEL: the HTTP/2 session died before the request was
|
||||
// sent, e.g. after the Bedrock SDK's 5-minute session timeout (#10379).
|
||||
"pending stream has been canceled",
|
||||
// Provider-requested retry delay cap failures should flow through the outer
|
||||
// retry policy so callers can surface/abort the backoff (#1123).
|
||||
"retry delay",
|
||||
// Explicit retry guidance emitted mid-stream by OpenAI Responses and Bedrock
|
||||
// stream exceptions (#6019).
|
||||
"you can retry your request",
|
||||
"try your request again",
|
||||
"please retry your request",
|
||||
// gRPC based providers (e.g. NVIDIA NIM)
|
||||
"ResourceExhausted",
|
||||
// Sign in with ChatGPT: usage or user data temporarily unavailable. Usage
|
||||
// failures can arrive mid-stream without an HTTP 503 in the message.
|
||||
"subscription_sharing_usage_unavailable",
|
||||
"subscription_sharing_user_unavailable",
|
||||
]);
|
||||
export function isRetryableAssistantError(message) {
|
||||
if (message.stopReason !== "error" || !message.errorMessage)
|
||||
return false;
|
||||
const errorMessage = message.errorMessage;
|
||||
if (NON_RETRYABLE_PROVIDER_LIMIT_ERROR_PATTERN.test(errorMessage))
|
||||
return false;
|
||||
return RETRYABLE_PROVIDER_ERROR_PATTERN.test(errorMessage);
|
||||
}
|
||||
import type { CommitPublication, Session } from '@earendil-works/pi-durable'
|
||||
|
||||
|
||||
export function onCommit(session: Session, advance: (publication: CommitPublication) => void) {
|
||||
const observation = session.observeCommits({ advance, end: () => {} })
|
||||
return () => observation.detach()
|
||||
}
|
||||
import type { AssistantMessage } from '@earendil-works/pi-ai'
|
||||
|
||||
|
||||
/**
|
||||
* Whether a model error is a rate limit Pi retries: HTTP 429, "rate limit", "too many requests". Retrying sooner only
|
||||
* hits the limit again, so Ultimate pauses until the user presses Play, and the held run continues through Pi's retry.
|
||||
* Quota, usage-limit and billing errors are on Pi's no-retry list: they end the run at once and fail as usual.
|
||||
* Overloaded servers (529) are the provider's capacity, not the account's limit.
|
||||
*/
|
||||
export const rateLimited = (error: string) =>
|
||||
/(?<![a-z0-9])429(?![a-z0-9])|rate.?limit|too many requests/i.test(error) &&
|
||||
isRetryableAssistantError({ stopReason: 'error', errorMessage: error } as AssistantMessage)
|
||||
@@ -0,0 +1,421 @@
|
||||
/** Durable provider-failure recovery and bounded reports through the messages service. */
|
||||
import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context'
|
||||
import { onCommit } from './retry.ts'
|
||||
import type { JsonValue } from '@earendil-works/chord'
|
||||
import type { AssistantMessage } from '@earendil-works/pi-ai'
|
||||
import { isRetryableAssistantError } from './retry.ts'
|
||||
import {
|
||||
defineDoc,
|
||||
defineExtension,
|
||||
defineTask,
|
||||
GenerationTask,
|
||||
hook,
|
||||
LiveDoc,
|
||||
ROOT_CONVERSATION_ID,
|
||||
} from '@earendil-works/pi-durable'
|
||||
import type {
|
||||
ConversationId,
|
||||
Cursor,
|
||||
EntryId,
|
||||
SubmissionRecord,
|
||||
TaskRecord,
|
||||
Tx,
|
||||
} from '@earendil-works/pi-durable'
|
||||
import type { Host as ExtensionHost } from '../../host/js/extensions.ts'
|
||||
import { rateLimited } from './retry.ts'
|
||||
import { MessagesApi, Messages } from '../messages/api.ts'
|
||||
import { Standing, Subagents } from '../subagents/api.ts'
|
||||
import { idleNote } from './idle.ts'
|
||||
import { WatchdogApi } from './api.ts'
|
||||
import type { ReportedFailure } from './api.ts'
|
||||
import type { Failure } from './refs.ts'
|
||||
|
||||
/** How long after a temporary failure its one retry starts. */
|
||||
const RETRY_AFTER = 60_000
|
||||
/** Reported failures kept for `failures()`; older ones drop off. */
|
||||
export const FAILURE_LIMIT = 20
|
||||
|
||||
const Reported = defineDoc<{ failures: ReportedFailure[] }>({
|
||||
kind: 'watchdog.reported',
|
||||
version: 1,
|
||||
scope: 'conversation',
|
||||
history: 'latest',
|
||||
fork: 'initial',
|
||||
checkpointWhen: () => false,
|
||||
initial: () => ({ failures: [] }),
|
||||
})
|
||||
|
||||
const Receipts = defineDoc<{ highWater?: number; tasks: Record<string, true> }>({
|
||||
kind: 'watchdog.receipts',
|
||||
version: 1,
|
||||
scope: 'conversation',
|
||||
history: 'latest',
|
||||
fork: 'initial',
|
||||
checkpointWhen: () => false,
|
||||
initial: () => ({ tasks: {} }),
|
||||
})
|
||||
const terminal = { status: 'terminal', outcome: { status: 'completed', result: null } } as const
|
||||
const failed = (task: TaskRecord<JsonValue, JsonValue, JsonValue>) =>
|
||||
task.kind === GenerationTask.definition.name &&
|
||||
task.state.status === 'terminal' &&
|
||||
task.state.outcome.status === 'failed' &&
|
||||
['model_error', 'no_model'].includes(
|
||||
(task.state.outcome.error.detail as { reason?: string } | undefined)?.reason ?? '',
|
||||
)
|
||||
const failedInput = (input: SubmissionRecord) =>
|
||||
input.type === 'input' &&
|
||||
input.status === 'unanswered' &&
|
||||
['model_error', 'no_model'].includes(input.reason)
|
||||
|
||||
/** The final error entry of a generation, excluding any later run. Pi can append several
|
||||
* assistant errors during its internal retries; only the last one ended the run. */
|
||||
async function boundary(tx: Tx, task: Generation, next?: Generation) {
|
||||
let upper = Number(task.id)
|
||||
let assistant: AssistantMessage | undefined
|
||||
if (
|
||||
task.state.status === 'terminal' &&
|
||||
task.state.outcome.status === 'failed' &&
|
||||
(task.state.outcome.error.detail as { reason?: string } | undefined)?.reason === 'model_error'
|
||||
) {
|
||||
let cursor: Cursor | undefined
|
||||
do {
|
||||
const page = await tx.scanEntries(
|
||||
{
|
||||
conversationId: task.conversationId,
|
||||
minEntryId: task.id as number as EntryId,
|
||||
...(next ? { maxEntryId: (next.id - 1) as EntryId } : {}),
|
||||
order: 'descending',
|
||||
},
|
||||
100,
|
||||
cursor,
|
||||
)
|
||||
const entry = page.items.find((candidate) => candidate.model?.[0]?.role === 'assistant')
|
||||
const message = entry?.model?.[0]
|
||||
if (message?.role === 'assistant') {
|
||||
if (message.stopReason === 'error') {
|
||||
upper = entry!.id
|
||||
assistant = message
|
||||
}
|
||||
break
|
||||
}
|
||||
cursor = page.next
|
||||
} while (cursor)
|
||||
}
|
||||
return { upper, assistant }
|
||||
}
|
||||
|
||||
/**
|
||||
* What a failed run's error asks for: a broken login needs the user, so it reports at once with a hint to log in again;
|
||||
* a rate limit that outlasted Pi's retries reports once and never re-prompts into the limit; an
|
||||
* incomplete Claude Code relay response (status 200 or unknown) is a dropped stream, so it retries like Pi's own
|
||||
* retryable errors.
|
||||
*/
|
||||
export function diagnose(error: string, assistant?: AssistantMessage) {
|
||||
if (rateLimited(error))
|
||||
return {
|
||||
hint: 'Still rate limited after Pi’s retries; once the limit resets (or the agent has another model), resume this conversation with a message',
|
||||
temporary: false,
|
||||
}
|
||||
if (/not logged in/i.test(error))
|
||||
return { hint: 'Run `claude auth login` in a terminal', temporary: false }
|
||||
// The conversation's model has no login here; its next input switches it to one that has.
|
||||
const missing = /Provider is not configured: ([\w-]+)|No API key (?:found )?for "?([\w-]+)/i.exec(
|
||||
error,
|
||||
)
|
||||
if (missing)
|
||||
return {
|
||||
hint: `${missing[1] ?? missing[2]} has no login here. Log in again (⌘K → Log in to a provider), or message the conversation: its next run switches to a model that has one`,
|
||||
temporary: false,
|
||||
}
|
||||
const provider = /OAuth refresh failed for ([\w-]+)/i.exec(error)?.[1] ?? assistant?.provider
|
||||
if (/OAuth refresh failed|invalid_grant/i.test(error))
|
||||
return {
|
||||
hint: `Log in to ${provider ?? 'the provider'} again: ⌘K → Log in to a provider`,
|
||||
temporary: false,
|
||||
}
|
||||
return {
|
||||
temporary:
|
||||
/^Incomplete Claude Code upstream response \(status (200|unknown),/.test(error) ||
|
||||
!!(assistant && isRetryableAssistantError(assistant)),
|
||||
}
|
||||
}
|
||||
|
||||
type Generation = TaskRecord<JsonValue, JsonValue, JsonValue>
|
||||
/** Live commits give us the exact settled inputs. On reopen, only model_error/no_model
|
||||
* submissions matter: their entries lie between consecutive failed-run boundaries.
|
||||
* Successful and stopped runs cannot match that filter. No-model has no assistant entry,
|
||||
* so its generation ID supplies its boundary. All IDs share one allocation order. */
|
||||
async function details(tx: Tx, metadata: ExtensionHost['metadata'], task: Generation) {
|
||||
const generations: Generation[] = []
|
||||
let cursor: Cursor | undefined
|
||||
do {
|
||||
const page = await tx.scanTasks(
|
||||
{ conversationId: task.conversationId, kind: GenerationTask.definition.name },
|
||||
500,
|
||||
cursor,
|
||||
)
|
||||
generations.push(...page.items)
|
||||
cursor = page.next
|
||||
} while (cursor)
|
||||
const index = generations.findIndex((candidate) => candidate.id === task.id)
|
||||
const previous = generations.findLastIndex(
|
||||
(candidate) => candidate.id < task.id && failed(candidate),
|
||||
)
|
||||
const { upper, assistant } = await boundary(tx, task, generations[index + 1])
|
||||
const lower =
|
||||
previous < 0 ? 0 : (await boundary(tx, generations[previous], generations[previous + 1])).upper
|
||||
let step = 'model request'
|
||||
cursor = undefined
|
||||
do {
|
||||
const page = await tx.scanEntries(
|
||||
{ conversationId: task.conversationId, maxEntryId: upper as EntryId },
|
||||
100,
|
||||
cursor,
|
||||
)
|
||||
for (const entry of page.items) {
|
||||
const message = entry.model?.[0]
|
||||
const call =
|
||||
message?.role === 'assistant' &&
|
||||
message.content.findLast((part) => part.type === 'toolCall')
|
||||
if (call && call.type === 'toolCall') {
|
||||
step = call.name
|
||||
break
|
||||
}
|
||||
}
|
||||
cursor = page.next
|
||||
} while (cursor && step === 'model request')
|
||||
const title = (await tx.doc(metadata, task.conversationId)).title ?? 'Untitled'
|
||||
return { upper, lower, step, assistant, title }
|
||||
}
|
||||
|
||||
export default function watchdog(host: ExtensionHost) {
|
||||
const Watchdog = defineTask<
|
||||
Failure & { failedTask: number; recipient: ConversationId; temporary: boolean; until: number },
|
||||
{ phase: 'wait' } | { phase: 'retry' } | { phase: 'report' },
|
||||
null
|
||||
>({
|
||||
name: 'watchdog.failure',
|
||||
version: 1,
|
||||
initial: (input) => ({ phase: input.temporary && !input.retried ? 'wait' : 'report' }),
|
||||
phases: {
|
||||
wait: (task, runtime, ctx) => runtime.sleepUntil(task.input.until, { phase: 'retry' }, ctx),
|
||||
retry: async (task, runtime, ctx) => {
|
||||
const { conversation, failedTask, error } = task.input
|
||||
const child = (await runtime.conversation(conversation as ConversationId, ctx))!
|
||||
// A replay after submit must finish, not mistake its own retry for someone else's recovery.
|
||||
let submitted = false
|
||||
let busy = false
|
||||
await runtime.commit(async (tx) => {
|
||||
submitted = !!(await tx.submissionByRequest(child.id, `watchdog-retry:${failedTask}`))
|
||||
busy = (await tx.doc(LiveDoc, child.id)).run !== undefined
|
||||
}, ctx)
|
||||
if (!submitted && !busy)
|
||||
await child.submit(
|
||||
{
|
||||
type: 'input',
|
||||
content: `Your previous run stopped with a provider error (${error}). Continue from where you left off; check what you already did and do not redo it.`,
|
||||
requestId: `watchdog-retry:${failedTask}`,
|
||||
whenBusy: 'followUp',
|
||||
},
|
||||
ctx,
|
||||
)
|
||||
await runtime.commit(() => terminal, ctx)
|
||||
},
|
||||
report: async (task, runtime, ctx) => {
|
||||
const { conversation, title, error, step, hint, retried, failedTask, recipient } =
|
||||
task.input
|
||||
const data: Failure = {
|
||||
conversation,
|
||||
title,
|
||||
error,
|
||||
step,
|
||||
retried,
|
||||
...(hint ? { hint } : {}),
|
||||
}
|
||||
const content = `[run failed] #${conversation} "${title}": ${error}. Last step: ${step}.${retried ? ' After one automatic retry.' : ''}${hint ? ` ${hint}.` : ''}`
|
||||
if (recipient === conversation) {
|
||||
// Notices are process-local diagnostics, not submissions; a reopened facet retains no old notice.
|
||||
host.notice(content)
|
||||
} else {
|
||||
await host.need(MessagesApi).send(
|
||||
conversation,
|
||||
{to: `#${recipient}`, text: `${content}\nCheck the conversation; resume it with a message, or tell the user if it needs them.`, reply: false},
|
||||
`watchdog-report:${failedTask}`,
|
||||
ctx,
|
||||
)
|
||||
}
|
||||
await runtime.commit(async (tx) => {
|
||||
const reported = await tx.doc(Reported, ROOT_CONVERSATION_ID)
|
||||
// Newest first by when the run failed: reports of older failures can finish later.
|
||||
reported.failures.push({ ...data, at: task.input.until - RETRY_AFTER })
|
||||
reported.failures.sort((a, b) => b.at - a.at)
|
||||
reported.failures.splice(FAILURE_LIMIT)
|
||||
// Reporting ends the watchdog's hold on a subagent's provider failure: its action has given up.
|
||||
const actions = Object.values((await tx.doc(Subagents, recipient)).actions)
|
||||
const action = actions.findLast((candidate) => candidate.conversationId === conversation)
|
||||
if (action?.status === 'failed') {
|
||||
const input = await tx.submissionByRequest(
|
||||
conversation as ConversationId,
|
||||
`subagent:${action.reporter}`,
|
||||
)
|
||||
if (input && failedInput(input)) action.status = 'given-up'
|
||||
}
|
||||
return terminal
|
||||
}, ctx)
|
||||
},
|
||||
},
|
||||
abort: (_task, runtime, ctx) =>
|
||||
runtime.commit(() => ({ status: 'terminal', outcome: { status: 'aborted' } }), ctx),
|
||||
})
|
||||
|
||||
async function detect(
|
||||
task: TaskRecord<JsonValue, JsonValue, JsonValue>,
|
||||
inputs?: SubmissionRecord[],
|
||||
) {
|
||||
if (!failed(task)) return
|
||||
let submissions = inputs
|
||||
if (!submissions) {
|
||||
submissions = []
|
||||
let cursor: Cursor | undefined
|
||||
do {
|
||||
const page = await host.storage.scanSubmissions(
|
||||
{ conversationId: task.conversationId, status: 'unanswered' },
|
||||
500,
|
||||
cursor,
|
||||
context,
|
||||
)
|
||||
submissions.push(...page.items.filter(failedInput))
|
||||
cursor = page.next
|
||||
} while (cursor)
|
||||
}
|
||||
await host.harness().commit(async (tx) => {
|
||||
const receipts = await tx.doc(Receipts, ROOT_CONVERSATION_ID)
|
||||
if (task.id <= (receipts.highWater ?? Infinity) || receipts.tasks[String(task.id)]) return
|
||||
const info = await details(tx, host.metadata, task)
|
||||
const runInputs = submissions.filter(
|
||||
(input) =>
|
||||
failedInput(input) &&
|
||||
input.entry !== undefined &&
|
||||
(inputs !== undefined || (input.entry > info.lower && input.entry <= info.upper)),
|
||||
)
|
||||
receipts.tasks[String(task.id)] = true
|
||||
const messages = (await tx.doc(Messages, ROOT_CONVERSATION_ID)).messages
|
||||
if (
|
||||
runInputs.some((input) => {
|
||||
const match = /^board:(\d+)$/.exec(input.requestId ?? '')
|
||||
return (
|
||||
match && messages.some((message) => message.id === Number(match[1]) && message.reply)
|
||||
)
|
||||
})
|
||||
)
|
||||
return
|
||||
const child = (await tx.conversation(task.conversationId))!
|
||||
const ultimate = (await tx.doc(Standing, ROOT_CONVERSATION_ID)).ultimate
|
||||
const recipient = child.owner?.conversationId ?? ultimate
|
||||
if (recipient === undefined)
|
||||
throw new Error('Watchdog needs @ultimate to report a failed run')
|
||||
if (task.state.status !== 'terminal' || task.state.outcome.status !== 'failed') return
|
||||
const error = task.state.outcome.error.message
|
||||
const { hint, temporary } = diagnose(error, info.assistant)
|
||||
await tx.createTask(
|
||||
Watchdog,
|
||||
{
|
||||
failedTask: task.id,
|
||||
conversation: child.id,
|
||||
recipient,
|
||||
title: info.title,
|
||||
error,
|
||||
step: info.step,
|
||||
...(hint ? { hint } : {}),
|
||||
retried: runInputs.some((input) => input.requestId?.startsWith('watchdog-retry:')),
|
||||
temporary,
|
||||
until: (task.endedAt ?? Date.now()) + RETRY_AFTER,
|
||||
},
|
||||
{ conversationId: recipient, ownership: { kind: 'conversation' }, background: true },
|
||||
)
|
||||
}, context)
|
||||
}
|
||||
|
||||
let unsubscribe: (() => void) | undefined
|
||||
let recovery = Promise.resolve()
|
||||
const close = async () => {
|
||||
unsubscribe?.()
|
||||
unsubscribe = undefined
|
||||
await recovery
|
||||
}
|
||||
host.onClose(close)
|
||||
host.implement(WatchdogApi, {
|
||||
failures: async (since) =>
|
||||
(
|
||||
(await host.harness().snapshot(Reported, ROOT_CONVERSATION_ID, context))?.failures ?? []
|
||||
).filter((failure) => failure.at >= since),
|
||||
})
|
||||
host.onOpen(async () => {
|
||||
unsubscribe?.()
|
||||
unsubscribe = onCommit(host.harness(), (publication) => {
|
||||
const inputs = publication.changes.flatMap((change) =>
|
||||
change.type === 'submission' && failedInput(change.value) ? [change.value] : [],
|
||||
)
|
||||
for (const change of publication.changes) {
|
||||
if (change.type !== 'task' || !failed(change.value)) continue
|
||||
recovery = recovery
|
||||
.then(() =>
|
||||
detect(
|
||||
change.value,
|
||||
inputs.filter((input) => input.conversationId === change.value.conversationId),
|
||||
),
|
||||
)
|
||||
.catch((error: unknown) => host.notice(`Watchdog recovery failed: ${String(error)}`))
|
||||
}
|
||||
})
|
||||
const tasks = await host.harness().commit(async (tx) => {
|
||||
const receipts = await tx.doc(Receipts, ROOT_CONVERSATION_ID)
|
||||
if (receipts.highWater === undefined) {
|
||||
let newest = 0
|
||||
let unfinished = Infinity
|
||||
let cursor: Cursor | undefined
|
||||
do {
|
||||
const page = await tx.scanTasks({}, 500, cursor)
|
||||
for (const task of page.items) {
|
||||
newest = task.id
|
||||
if (task.state.status !== 'terminal') unfinished = Math.min(unfinished, task.id)
|
||||
// Terminal failures are historical even when a still-running older task holds the mark back.
|
||||
if (failed(task)) receipts.tasks[String(task.id)] = true
|
||||
}
|
||||
cursor = page.next
|
||||
} while (cursor)
|
||||
receipts.highWater = unfinished === Infinity ? newest : unfinished - 1
|
||||
return []
|
||||
}
|
||||
const all = []
|
||||
let cursor: Cursor | undefined
|
||||
do {
|
||||
const page = await tx.scanTasks({ order: 'descending' }, 500, cursor)
|
||||
all.push(...page.items.filter((task) => task.id > receipts.highWater!))
|
||||
cursor = (page.items.at(-1)?.id ?? 0) > receipts.highWater ? page.next : undefined
|
||||
} while (cursor)
|
||||
return all.toReversed()
|
||||
}, context)
|
||||
for (const task of tasks) await detect(task)
|
||||
// Do not pass unfinished tasks: an older run can fail after a newer one, even across restart.
|
||||
await host.harness().commit(async (tx) => {
|
||||
const receipts = await tx.doc(Receipts, ROOT_CONVERSATION_ID)
|
||||
for (const task of tasks) {
|
||||
if (task.state.status !== 'terminal') break
|
||||
receipts.highWater = task.id
|
||||
}
|
||||
}, context)
|
||||
})
|
||||
return defineExtension({
|
||||
name: 'watchdog',
|
||||
tasks: [Watchdog],
|
||||
hooks: [
|
||||
hook(GenerationTask, {
|
||||
beforeInput: async (_input, { tx, conversationId }) =>
|
||||
(await tx.doc(Standing, ROOT_CONVERSATION_ID)).ultimate === conversationId
|
||||
? idleNote(tx, host, conversationId, Date.now())
|
||||
: undefined,
|
||||
}),
|
||||
],
|
||||
})
|
||||
}
|
||||
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"scripts": { "bundle": "nu apply-patch.nu && esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js && esbuild foundation-test.ts --bundle --format=esm --platform=browser --outfile=foundation-test-bundle.js && esbuild extensions-test.ts --bundle --format=esm --platform=browser --outfile=extensions-test-bundle.js && esbuild events-test.ts --bundle --format=esm --platform=browser --outfile=events-test-bundle.js && esbuild subagents-test.ts --bundle --format=esm --platform=browser --outfile=subagents-test-bundle.js && esbuild core-ext-test.ts --bundle --format=esm --platform=browser --outfile=core-ext-test-bundle.js && esbuild plan-todos-test.ts --bundle --format=esm --platform=browser --outfile=plan-todos-test-bundle.js && esbuild p2-test.ts --bundle --format=esm --platform=browser --outfile=p2-test-bundle.js && esbuild chain-test.ts --bundle --format=esm --platform=browser --outfile=chain-test-bundle.js && esbuild undo-harness-test.ts --bundle --format=esm --platform=browser --outfile=undo-harness-test-bundle.js && esbuild messages-test.ts --bundle --format=esm --platform=browser --outfile=messages-test-bundle.js && esbuild p3-test.ts --bundle --format=esm --platform=browser --outfile=p3-test-bundle.js && esbuild automations-test.ts --bundle --format=esm --platform=browser --outfile=automations-test-bundle.js" },
|
||||
"scripts": { "bundle": "nu apply-patch.nu && esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js && esbuild foundation-test.ts --bundle --format=esm --platform=browser --outfile=foundation-test-bundle.js && esbuild extensions-test.ts --bundle --format=esm --platform=browser --outfile=extensions-test-bundle.js && esbuild events-test.ts --bundle --format=esm --platform=browser --outfile=events-test-bundle.js && esbuild subagents-test.ts --bundle --format=esm --platform=browser --outfile=subagents-test-bundle.js && esbuild core-ext-test.ts --bundle --format=esm --platform=browser --outfile=core-ext-test-bundle.js && esbuild plan-todos-test.ts --bundle --format=esm --platform=browser --outfile=plan-todos-test-bundle.js && esbuild p2-test.ts --bundle --format=esm --platform=browser --outfile=p2-test-bundle.js && esbuild chain-test.ts --bundle --format=esm --platform=browser --outfile=chain-test-bundle.js && esbuild undo-harness-test.ts --bundle --format=esm --platform=browser --outfile=undo-harness-test-bundle.js && esbuild messages-test.ts --bundle --format=esm --platform=browser --outfile=messages-test-bundle.js && esbuild p3-test.ts --bundle --format=esm --platform=browser --outfile=p3-test-bundle.js && esbuild automations-test.ts --bundle --format=esm --platform=browser --outfile=automations-test-bundle.js && esbuild watchdog-test.ts --bundle --format=esm --platform=browser --outfile=watchdog-test-bundle.js" },
|
||||
"dependencies": {
|
||||
"@earendil-works/pi-durable": "1.5.0",
|
||||
"@earendil-works/pi-ai": "1.5.0",
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,105 @@
|
||||
/** World-loaded watchdog retries, message reports, and recovery with a virtual clock. */
|
||||
import {BACKGROUND_CONTEXT as context} from '@earendil-works/chord/context';
|
||||
import {createModels} from '@earendil-works/pi-ai/models';
|
||||
import {fauxProvider, fauxAssistantMessage} from '@earendil-works/pi-ai/providers/faux';
|
||||
import {Harness, createRegistry, configure, LiveDoc} from '@earendil-works/pi-durable';
|
||||
import {loadExtensions, fs, Metadata} from './extensions';
|
||||
import {openStorage} from './sqlite';
|
||||
const assert = (value: any, message: string) => {if (!value) throw new Error(message)};
|
||||
const until = async (test: () => any) => {const deadline = Date.now() + 15000; while (!await test()) {if (Date.now() > deadline) throw new Error('Watchdog deadline'); await new Promise(r => setTimeout(r, 0))}};
|
||||
let now = globalThis.watchdogMode === 'resume' ? 1_060_000 : 1_000_000;
|
||||
const timers = new Set<any>();
|
||||
const models = createModels(), faux = fauxProvider({tokensPerSecond: 100000}); models.setProvider(faux.provider);
|
||||
const registry = createRegistry(), storage = await openStorage(globalThis.storagePath);
|
||||
const harness = await Harness.open(storage, {models, registry, now: () => now, setTimer: (callback, delay) => {const timer = {callback, at: now + delay}; timers.add(timer); return () => timers.delete(timer)}, settings: {retry: {maxRetries: 0}, get extensions() {return registry.snapshot().installed()}}}, context);
|
||||
const root = await harness.root(context, {agent: {model: {provider: 'faux', modelId: 'faux-1'}, cwd: '/src/sausage'}});
|
||||
const route = (transcript: any): any => {
|
||||
faux.appendResponses([route]);
|
||||
const last = transcript.messages.findLast((m: any) => m.role === 'user');
|
||||
const text = typeof last?.content === 'string' ? last.content : '';
|
||||
if (text === 'fail' || text.startsWith('Your previous run')) return fauxAssistantMessage('', {stopReason: 'error', errorMessage: 'Our servers are currently overloaded. Please try again later.'});
|
||||
if (text === 'permanent') return fauxAssistantMessage('', {stopReason: 'error', errorMessage: 'invalid model configuration'});
|
||||
return fauxAssistantMessage('ok');
|
||||
};
|
||||
faux.setResponses([route]);
|
||||
const loader = await loadExtensions(harness, registry, storage, models, context);
|
||||
assert(!loader.state().error, `Extension load: ${loader.state().error}`);
|
||||
const sha = await fs.read('/src/sausage/.git/sausage-last-good');
|
||||
const base = `file:///run/sausage/extensions/${sha}/extensions`;
|
||||
const {Standing} = await import(`${base}/subagents/api.ts`);
|
||||
const {Messages} = await import(`${base}/messages/api.ts`);
|
||||
const {diagnose} = await import(`${base}/watchdog/server.ts`);
|
||||
const service = globalThis.watchdogService;
|
||||
const tasks = async () => (await harness.commit((tx: any) => tx.scanTasks({kind: 'watchdog.failure'}, 100), context)).items;
|
||||
const records = async () => (await harness.commit((tx: any) => tx.scanConversations({}, 100), context)).items;
|
||||
const create = async (title: string) => harness.commit(async (tx: any) => {
|
||||
const record = await tx.createConversation({ownership: {kind: 'ownerless'}});
|
||||
await configure(tx, record.id, {model: {provider: 'faux', modelId: 'faux-1'}, cwd: '/src/sausage'});
|
||||
Object.assign(await tx.doc(Metadata, record.id), {title});
|
||||
return record.id;
|
||||
}, context);
|
||||
let primary = (await harness.snapshot(Standing, root.id, context)).ultimate;
|
||||
if (!primary) {primary = await create('@ultimate'); await harness.commit(async (tx: any) => {(await tx.doc(Standing, root.id)).ultimate = primary}, context)}
|
||||
const submit = async (id: number, text: string, requestId?: string) => (await (await harness.conversation(id, context))!.submit({type: 'input', content: text, requestId}, context)).wait(context);
|
||||
const requests = async (id: number) => (await storage.scanSubmissions({conversationId: id}, 100, undefined, context)).items;
|
||||
const retries = async (id: number) => (await requests(id)).filter((r: any) => r.requestId?.startsWith('watchdog-retry:'));
|
||||
const reports = async (id: number) => (await harness.snapshot(Messages, root.id, context)).messages.filter((m: any) => m.from === id && m.text.startsWith('[run failed]'));
|
||||
const advance = (ms: number) => {now += ms; for (const timer of [...timers]) if (timer.at <= now) {timers.delete(timer); timer.callback()} harness.resume()};
|
||||
harness.resume();
|
||||
try {
|
||||
if (globalThis.watchdogMode === 'resume') {
|
||||
const ids = await records();
|
||||
const branch = (await Promise.all(ids.map(async (r: any) => ({id: r.id, title: (await harness.snapshot(Metadata, r.id, context))?.title})))).find(r => r.title === 'Crash gap')!.id;
|
||||
await until(async () => (await reports(branch)).length === 1);
|
||||
await harness.waitForIdle(context);
|
||||
assert((await retries(branch)).length === 1, 'Restart duplicated the retry');
|
||||
assert((await reports(branch)).length === 1, 'Restart duplicated the report');
|
||||
await until(async () => (await service.failures(0)).filter((f: any) => f.conversation === branch).length === 1);
|
||||
assert((await service.failures(0)).filter((f: any) => f.conversation === branch).length === 1, 'Restart duplicated failure history');
|
||||
} else {
|
||||
const branch = await create(globalThis.watchdogMode === 'kill' ? 'Crash gap' : 'Provider blip');
|
||||
await submit(branch, 'fail');
|
||||
await until(async () => (await tasks()).some((t: any) => t.state.status === 'sleeping'));
|
||||
const sleeper = (await tasks()).find((t: any) => t.state.status === 'sleeping');
|
||||
assert(sleeper.state.until === 1_060_000, 'Retry deadline must be durable');
|
||||
assert((await retries(branch)).length === 0 && (await reports(branch)).length === 0, 'Premature retry or report');
|
||||
if (globalThis.watchdogMode === 'kill') {await fs.write('/watchdog-ready', 'yes'); await new Promise(() => {})}
|
||||
advance(59_999); await harness.waitForIdle(context);
|
||||
assert((await retries(branch)).length === 0, 'Retry happened before its deadline');
|
||||
advance(1);
|
||||
await until(async () => (await reports(branch)).length === 1);
|
||||
await harness.waitForIdle(context);
|
||||
assert((await retries(branch)).length === 1, 'Expected one retry');
|
||||
assert((await reports(branch))[0].to === primary, 'The message must reach @ultimate');
|
||||
assert((await reports(branch))[0].text.includes('After one automatic retry'), 'Report must describe its retry');
|
||||
await until(async () => (await service.failures(0)).length === 1);
|
||||
const history = await service.failures(0);
|
||||
assert(history.length === 1 && history[0].retried && history[0].title === 'Provider blip', 'Failure history');
|
||||
advance(120_000); await harness.waitForIdle(context);
|
||||
assert((await retries(branch)).length === 1 && (await reports(branch)).length === 1, 'A failed retry must not retry again');
|
||||
const permanent = await create('Bad config'); await submit(permanent, 'permanent');
|
||||
await until(async () => (await reports(permanent)).length === 1);
|
||||
assert((await retries(permanent)).length === 0, 'Permanent errors must report without retry');
|
||||
const excluded = await create('Board owns failure');
|
||||
await globalThis.messageService.send(primary, {to: `#${excluded}`, text: 'healthy', reply: false}, 'setup', context);
|
||||
await harness.waitForIdle(context);
|
||||
await harness.commit(async (tx: any) => {(await tx.doc(Messages, root.id)).messages.push({id: 999, from: primary, to: excluded, text: 'fail', at: now, reply: true, state: 'delivered'})}, context);
|
||||
await submit(excluded, 'fail', 'board:999'); await harness.waitForIdle(context);
|
||||
assert(!(await tasks()).some((t: any) => t.input.conversation === excluded), 'Reply-expected messages handle their own failures');
|
||||
for (const [text, temporary] of [['Overloaded (529)', true], ['429 Too Many Requests', false], ['invalid model configuration', false], ['Incomplete Claude Code upstream response (status unknown, capture incomplete, relay failure Error)', true]] as const) {
|
||||
assert(diagnose(text, fauxAssistantMessage('', {stopReason: 'error', errorMessage: text})).temporary === temporary, `Diagnosis: ${text}`);
|
||||
}
|
||||
assert(!(await harness.snapshot(LiveDoc, branch, context)).run, 'Failed generation must stop');
|
||||
const {idleNote} = await import(`${base}/watchdog/idle.ts`);
|
||||
const {Branch} = await import(`${base}/lineage/api.ts`);
|
||||
await harness.commit(async (tx: any) => {
|
||||
(await tx.doc(Branch, branch)).branch = {from: root.id, at: 0, intent: 'Idle work'};
|
||||
Object.assign(await tx.doc(Metadata, branch), {inputAt: 1, answerAt: 1});
|
||||
}, context);
|
||||
const idle = () => harness.commit((tx: any) => idleNote(tx, globalThis.testHost, primary, 45 * 60_000), context);
|
||||
assert((await idle()).entries[0].data.ids.includes(branch), 'Idle branch reminder');
|
||||
assert(await idle() === undefined, 'Unchanged idle IDs must not repeat');
|
||||
await harness.commit(async (tx: any) => {delete (await tx.doc(Branch, branch)).branch}, context);
|
||||
assert((await idle()).entries[0].model[0].content === 'No branches are idle.', 'Idle reminder must clear');
|
||||
}
|
||||
} finally {await loader.close(); await harness.close(context)}
|
||||
@@ -0,0 +1,190 @@
|
||||
//! World-loaded watchdog recovery with SIGKILL during its durable wait.
|
||||
use sausage::{harness::run_module, world::World};
|
||||
use std::{
|
||||
path::Path,
|
||||
process::Command,
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
const BUNDLE: &str = include_str!("../js/watchdog-test-bundle.js");
|
||||
static PREPARE: std::sync::Mutex<()> = std::sync::Mutex::new(());
|
||||
fn prepare(dir: &Path) -> anyhow::Result<()> {
|
||||
let _guard = PREPARE.lock().unwrap();
|
||||
let disk = dir.join("world");
|
||||
let repo = disk.join("src/sausage");
|
||||
if repo.join(".git").exists() {
|
||||
return Ok(());
|
||||
}
|
||||
let source = Path::new(env!("CARGO_MANIFEST_DIR")).join("..");
|
||||
let target = Path::new(env!("CARGO_TARGET_TMPDIR")).parent().unwrap();
|
||||
let userland = source.join("userland");
|
||||
if !userland.join("cwd-init/init.o").exists() {
|
||||
assert!(
|
||||
Command::new("nu")
|
||||
.args(["--no-config-file", "cwd-init/build.nu"])
|
||||
.current_dir(&userland)
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
}
|
||||
assert!(
|
||||
Command::new("cargo")
|
||||
.args(["build", "--release", "-p", "git", "--manifest-path"])
|
||||
.arg(userland.join("Cargo.toml"))
|
||||
.args(["--target", "wasm32-wasip2", "--target-dir"])
|
||||
.arg(target)
|
||||
.args(["-j", "2", "--locked"])
|
||||
.current_dir(&userland)
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
std::fs::create_dir_all(disk.join("bin"))?;
|
||||
std::fs::copy(
|
||||
target.join("wasm32-wasip2/release/git.wasm"),
|
||||
disk.join("bin/git.wasm"),
|
||||
)?;
|
||||
for directory in [
|
||||
"extensions/subagents",
|
||||
"extensions/messages",
|
||||
"extensions/watchdog",
|
||||
"extensions/lineage",
|
||||
"extensions/core",
|
||||
"extensions/conversations",
|
||||
"extensions/models",
|
||||
"agents",
|
||||
] {
|
||||
std::fs::create_dir_all(repo.join(directory))?;
|
||||
}
|
||||
for directory in [
|
||||
"extensions/subagents",
|
||||
"extensions/messages",
|
||||
"extensions/watchdog",
|
||||
"extensions/lineage",
|
||||
"extensions/models",
|
||||
] {
|
||||
for entry in std::fs::read_dir(source.join(directory))? {
|
||||
let entry = entry?;
|
||||
if entry.path().extension().is_some_and(|ext| ext == "ts") {
|
||||
std::fs::copy(entry.path(), repo.join(directory).join(entry.file_name()))?;
|
||||
}
|
||||
}
|
||||
}
|
||||
std::fs::create_dir_all(repo.join("extensions/probe"))?;
|
||||
std::fs::write(
|
||||
repo.join("extensions/probe/server.ts"),
|
||||
"export default host => { globalThis.testHost = host; globalThis.subagentService = host.need({name:'subagents'}); globalThis.messageService = host.need({name:'messages'}); globalThis.watchdogService = host.need({name:'watchdog'}); globalThis.lineageService = host.need({name:'lineage'}); host.implement({name:'conversations'}, {summaryModel: async () => globalThis.summaryRef, concludedIn: async () => new Set(), generatedTitle: async () => {}, setStanding: async () => {}, setAutomaticSummaries: async () => {}}); return {name:'probe'}; }",
|
||||
)?;
|
||||
std::fs::copy(
|
||||
source.join("extensions/core/agents.ts"),
|
||||
repo.join("extensions/core/agents.ts"),
|
||||
)?;
|
||||
std::fs::copy(
|
||||
source.join("extensions/conversations/api.ts"),
|
||||
repo.join("extensions/conversations/api.ts"),
|
||||
)?;
|
||||
std::fs::write(
|
||||
repo.join("agents/models.json"),
|
||||
r#"{"default":{"model":"faux/faux-1"}}"#,
|
||||
)?;
|
||||
std::fs::write(
|
||||
repo.join("agents/scout.md"),
|
||||
"---\nname: scout\ndescription: Search\nresident: true\ndenyTools: [subagent]\nisolate: none\n---\nScout instructions\n",
|
||||
)?;
|
||||
std::fs::write(
|
||||
repo.join("agents/ultimate.md"),
|
||||
"---\nname: ultimate\ndescription: Primary\nresident: true\nisolate: none\n---\nPrimary agent\n",
|
||||
)?;
|
||||
std::fs::write(
|
||||
repo.join("agents/coordinator.md"),
|
||||
"---\nname: coordinator\ndescription: Coordinate\nisolate: none\n---\nCoordinator agent\n",
|
||||
)?;
|
||||
for args in [
|
||||
vec!["init", "-b", "main"],
|
||||
vec!["add", "."],
|
||||
vec!["commit", "--no-gpg-sign", "-m", "watchdog fixture"],
|
||||
] {
|
||||
assert!(
|
||||
Command::new("git")
|
||||
.args(args)
|
||||
.current_dir(&repo)
|
||||
.env("GIT_AUTHOR_NAME", "Test")
|
||||
.env("GIT_AUTHOR_EMAIL", "test@example.com")
|
||||
.env("GIT_COMMITTER_NAME", "Test")
|
||||
.env("GIT_COMMITTER_EMAIL", "test@example.com")
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
async fn run(dir: &Path, mode: &str) -> anyhow::Result<()> {
|
||||
let world = World::open(dir)?;
|
||||
prepare(dir)?;
|
||||
let exec = sausage::process::exec(World {
|
||||
root: world.root.clone(),
|
||||
host_dir: world.host_dir.clone(),
|
||||
processes: world.processes.clone(),
|
||||
})?;
|
||||
run_module(
|
||||
world,
|
||||
exec,
|
||||
&format!("globalThis.watchdogMode={mode:?};\n{BUNDLE}"),
|
||||
)
|
||||
.await
|
||||
}
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn coordination() -> anyhow::Result<()> {
|
||||
let dir = tempfile::tempdir()?;
|
||||
run(dir.path(), "normal").await
|
||||
}
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn restart_child() -> anyhow::Result<()> {
|
||||
if let Ok(dir) = std::env::var("WATCHDOG_DIR") {
|
||||
run(Path::new(&dir), &std::env::var("WATCHDOG_MODE")?).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
#[test]
|
||||
fn sigkill_during_durable_wait() -> anyhow::Result<()> {
|
||||
let dir = tempfile::tempdir()?;
|
||||
World::open(dir.path())?;
|
||||
prepare(dir.path())?;
|
||||
let spawn = |mode: &str| {
|
||||
Command::new(std::env::current_exe().unwrap())
|
||||
.args(["--exact", "restart_child", "--nocapture"])
|
||||
.env("WATCHDOG_DIR", dir.path())
|
||||
.env("WATCHDOG_MODE", mode)
|
||||
.spawn()
|
||||
};
|
||||
let mut child = spawn("kill")?;
|
||||
let deadline = Instant::now() + Duration::from_secs(30);
|
||||
while !dir.path().join("world/watchdog-ready").exists() {
|
||||
if let Some(status) = child.try_wait()? {
|
||||
anyhow::bail!("Child exited before reply: {status}");
|
||||
}
|
||||
if Instant::now() > deadline {
|
||||
child.kill()?;
|
||||
child.wait()?;
|
||||
anyhow::bail!("Delivery did not reach reply");
|
||||
}
|
||||
std::thread::yield_now();
|
||||
}
|
||||
child.kill()?;
|
||||
child.wait()?;
|
||||
for _ in 0..2 {
|
||||
let mut child = spawn("resume")?;
|
||||
let deadline = Instant::now() + Duration::from_secs(30);
|
||||
loop {
|
||||
if let Some(status) = child.try_wait()? {
|
||||
assert!(status.success());
|
||||
break;
|
||||
}
|
||||
if Instant::now() > deadline {
|
||||
child.kill()?;
|
||||
child.wait()?;
|
||||
anyhow::bail!("Reply recovery timed out");
|
||||
}
|
||||
std::thread::yield_now();
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user