vrr/CodeNomad
2
0
Fork 0
mirror of https://github.com/NeuralNomadsAI/CodeNomad.git synced 2026-10-21 05:06:37 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

test(missions): await recurring fixture clock quiescence (#1044)

Browse source 
## Cause
Remaining recurring-day CI failures observe a coordinator before its
awaited admission/error publication. The fixture clock advance
previously waited four timer ticks, not a completed clock turn.
Production recurrence correctly awaits those operations.

## Fix
Track each Job's own live sleep or terminal exit and await bounded
quiescence on each wake. Stale race sleepers no longer mask active work.
Existing five-second deadline and parked model/Form semantics retained;
no production recurrence changes.

## Evidence
Held-admission success/refusal regressions fail with original advance
and pass fixed. Four fixture callers36/36; recurring-day with
controlled25ms and75ms admission delays25/25 each; clock units7/7;
server typecheck and diff check pass. No assertion
weakening/skips/timeout increases. Full server CI not rerun locally. All
fixtures isolated; no shared daemon/user DB.

Oversized touched source: recurring-day-fixture.ts ~560 lines; test452
lines.
This commit is contained in:
Pascal André 2026-10-20 20:02:22 +02:00 • committed by GitHub
parent 04e7aae1db
commit 4d4bf14908
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 65 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

49
packages/server/src/opencode/missions/recurring-day-fixture.ts
View file

@ -1,4 +1,5 @@
import assert from "node:assert/strict"
import { AsyncLocalStorage } from "node:async_hooks"
import { execFileSync } from "node:child_process"
import { createHash, createHmac } from "node:crypto"
import { realpathSync } from "node:fs"
@ -29,7 +30,8 @@ export const START = Date.parse("2026-10-08T00:00:00Z")
export const DUE = START + (8 * 60 + 15) * 60_000
type Handler = (input: unknown, context: unknown) => Effect.Effect<unknown, unknown>
type NativeTool = { name: string; options?: { namespace?: string }; execute: Handler }
type Job = { id: string; type: string; status: string; metadata?: Record<string, unknown>; run: Effect.Effect<string, unknown>; fiber?: Fiber.Fiber<string, unknown> }
type Sleeper = { at: number; wake: () => void }
type Job = { id: string; type: string; status: string; metadata?: Record<string, unknown>; run: Effect.Effect<string, unknown>; fiber?: Fiber.Fiber<string, unknown>; parked?: Sleeper }
type SessionInfo = typeof Session.Info.Type
type MessageInfo = typeof SessionMessage.Info.Type
export type Schedule = {
@ -61,6 +63,8 @@ export class RecurringDayFixture {
failScheduleJobRead: (() => boolean) | undefined
/** Fresh native first admissions (a re-admitted ID after pruning counts again). */
admissions = 0
/** Controlled native I/O boundary; tests may hold admission after coordinator creation. */
beforeAdmission: (() => Promise<void>) | undefined
readonly tools = new Map<string, NativeTool>()
readonly calls: Array<{ name: string; sessionID: string }> = []
readonly jobs = new Map<string, Job>()
@ -71,7 +75,9 @@ export class RecurringDayFixture {
readonly errors: unknown[] = []
private handlers = new Map<string, Record<string, Handler>>()
private hooks: Array<(event: unknown) => Effect.Effect<unknown, unknown>> = []
private sleepers = new Set<{ at: number; wake: () => void }>()
private sleepers = new Set<Sleeper>()
// Effect preserves Node async context across Job/race fibers, even when it replaces its service context.
private jobContext = new AsyncLocalStorage<Job>()
private busListeners = new Set<(event: unknown) => Effect.Effect<void>>()
private scope!: Scope.Closeable
private graph!: Context.Context<never>
@ -190,11 +196,15 @@ export class RecurringDayFixture {
const entries = rows.slice(0, limit).map(row => ({ key: row.key.slice(this.key("").length), value: JSON.parse(row.value) }))
return { entries, ...(rows.length > limit ? { next: entries.at(-1)!.key } : {}) }
}) }
const admit = (input: Record<string, unknown>, type: "user" | "synthetic") => Effect.promise(async () => {
await this.beforeAdmission?.()
return this.admit(input, type)
})
const native = {
get: (id: string) => Effect.sync(() => this.session(id)),
create: (input: Record<string, unknown>) => Effect.sync(() => this.createSession(input)),
prompt: (input: Record<string, unknown>) => Effect.sync(() => this.admit(input, "user")),
synthetic: (input: Record<string, unknown>) => Effect.sync(() => this.admit(input, "synthetic")),
prompt: (input: Record<string, unknown>) => admit(input, "user"),
synthetic: (input: Record<string, unknown>) => admit(input, "synthetic"),
message: (input: { messageID: string }) => Effect.sync(() => this.message(input.messageID)),
messages: (input: { sessionID: string }) => Effect.sync(() => this.messages(input.sessionID)),
inbox: (id: string) => Effect.sync(() => this.db.prepare("SELECT payload FROM session_inbox WHERE session_id=?").all(id)
@ -212,10 +222,10 @@ export class RecurringDayFixture {
if (this.jobStartFailure === "absent") return Effect.die(new Error("Injected native Job start failure"))
const old = this.jobs.get(input.id)
if (old?.status === "running") return Effect.succeed(old)
const record = { ...input, status: "running" }
const record: Job = { ...input, status: "running", parked: undefined }
this.jobs.set(input.id, record)
record.fiber = Effect.runForkWith(this.graph)(input.run.pipe(Effect.provideService(Clock.Clock, this.clock)))
void Effect.runPromise(Fiber.await(record.fiber)).then(exit => {
record.fiber = this.jobContext.run(record, () => Effect.runForkWith(this.graph)(input.run.pipe(Effect.provideService(Clock.Clock, this.clock))))
record.fiber.addObserver(exit => {
record.status = Exit.isSuccess(exit) ? "completed" : "error"
if (Exit.isFailure(exit)) this.errors.push(exit.cause)
})
@ -281,7 +291,10 @@ export class RecurringDayFixture {
currentTimeNanosUnsafe: () => BigInt(this.now) * 1_000_000n, currentTimeNanos: Effect.sync(() => BigInt(this.now) * 1_000_000n),
monotonicTimeNanosUnsafe: () => BigInt(this.now) * 1_000_000n, monotonicTimeNanos: Effect.sync(() => BigInt(this.now) * 1_000_000n),
sleep: duration => Effect.callback<void>(resume => {
const job = this.jobContext.getStore()
assert(job, "virtual sleep must belong to a native fixture Job")
const sleeper = { at: this.now + Duration.toMillis(duration), wake: () => {
job.parked = undefined
// The journey spans 32+ hours; measure the requested per-virtual-day
// ceiling, not its total duration. Keep the maximum monotonic for Pause.
this.wakeTimes = this.wakeTimes.filter(at => at > this.now - DAY)
@ -290,7 +303,11 @@ export class RecurringDayFixture {
resume(Effect.void)
} }
this.sleepers.add(sleeper)
return Effect.sync(() => { this.sleepers.delete(sleeper) })
job.parked = sleeper
return Effect.sync(() => {
this.sleepers.delete(sleeper)
if (job.parked === sleeper) job.parked = undefined
})
}),
}
async flush() { for (let n = 0; n < 4; n++) await new Promise<void>(resolve => setTimeout(resolve, 1)) }
@ -300,26 +317,28 @@ export class RecurringDayFixture {
const deadline = performance.now() + timeoutMs
while (!await predicate()) {
assert(performance.now() < deadline, message)
await this.flush()
await new Promise<void>(resolve => setImmediate(resolve))
}
}
/** Every running Job is parked in a virtual sleep, so no background write is in flight. */
/** Wait for each Job's next sleep or exit, not an unrelated/stale race sleeper.
* Native model/Form waits are session state; they do not keep a clock turn in flight. */
async quiescent(message = "background Jobs settled") {
await this.until(() => [...this.jobs.values()].filter(job => job.status === "running").length <= this.sleepers.size, message)
await this.until(() => [...this.jobs.values()].every(job => job.status !== "running"
|| job.parked !== undefined && this.sleepers.has(job.parked)), message)
}
async advance(to: number) {
assert(to >= this.now)
await this.flush()
await this.quiescent()
while (true) {
const next = Math.min(...[...this.sleepers].map(s => s.at))
if (next > to) break
this.now = next
for (const sleeper of [...this.sleepers]) if (sleeper.at <= next) { this.sleepers.delete(sleeper); sleeper.wake() }
await this.flush()
for (const sleeper of [...this.sleepers]) if (sleeper.at <= next && this.sleepers.delete(sleeper)) sleeper.wake()
await this.quiescent()
assert(this.wakeups < 3000, "virtual day runner is spinning")
}
this.now = to
await this.flush()
await this.quiescent()
}
private fail(point: NonNullable<RecurringDayFixture["crash"]>) {
if (this.crash !== point) return

31
packages/server/src/opencode/missions/recurring-day.e2e.test.ts
View file

@ -8,6 +8,37 @@ import { DAY, DUE, RecurringDayFixture } from "./recurring-day-fixture"
// All model-produced business state goes through registered real mission_* tools.
// The fixture never appends journal events, fabricates final reports or calls recurrence reserve/finish.
test("Clock advance waits for the woken Job's admission and error publication, not timer ticks", async t => {
for (const fail of [false, true]) {
await t.test(fail ? "admission refusal" : "admission success", async () => {
const f = await RecurringDayFixture.open()
let release!: () => void
const held = new Promise<void>(resolve => { release = resolve })
let entered = false, returned = false
let advancing: Promise<void> | undefined
try {
await f.create(); await f.control("play")
if (fail) { f.crash = "before-message"; await f.advance(DUE); f.crash = "before-message" }
f.beforeAdmission = async () => { entered = true; await held }
advancing = f.advance(fail ? DUE + 3_600_000 : DUE).then(() => { returned = true })
await f.until(() => entered, "native admission reached")
// Drain the old fixed flush budget while the native callback is still held.
await f.flush(); await f.flush()
assert.equal(f.coordinators.length, 1)
assert.equal(f.starts.length, 0)
assert.equal(returned, false, "advance must not return between native create and admission")
release(); await advancing
assert.equal(f.starts.length, fail ? 0 : 1)
const settledTurn = await f.snapshot()
assert.equal(settledTurn.lastError?.code, fail ? "admission-failed" : undefined)
assert.equal(settledTurn.pending?.status, fail ? "uncertain" : "running")
// An active coordinator/Form is a valid parked Job, not an in-flight clock turn.
if (!fail) { f.pendingForm(); await f.advance(DUE + 3_600_000); assert((await f.snapshot()).pending) }
} finally { try { release(); await advancing } finally { await f.close() } }
})
}
})
test("A. Full day: paused create, due work, real tools, archive and a distinct next day", async t => {
const f = await RecurringDayFixture.open(); t.after(() => f.close())
const created = await f.create()