Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions packages/opencode/src/goal/goal.ts
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,11 @@ export interface Interface {
* goalID+revision pair. Required (not optional): an omitted expected would
* let a stale loop result mutate whatever goal replaced the judged one. */
expected: { readonly goalID: string; readonly revision: number },
/** issue #285: the assistant message ID judged for this evaluation.
* Persisted on continue commits as the DURABLE crash-recovery gate — the
* boot scan skips a window still ending on this boundary (the
* process-local evaluatedRevisions map cannot survive a crash). */
judged?: string,
) => Effect.Effect<
| {
state: GoalState.Info
Expand Down Expand Up @@ -678,6 +683,7 @@ const serviceLayer = Layer.effect(
reason: string,
parseFailed: boolean,
expected: { readonly goalID: string; readonly revision: number },
judged?: string,
) {
return yield* transition(sessionID, (state) => {
if (!state || state.status !== "active" || !matchesExpected(state, expected))
Expand Down Expand Up @@ -739,6 +745,8 @@ const serviceLayer = Layer.effect(
last_reason: reason,
paused_reason: pauseReason,
consecutive_parse_failures: GoalState.nni(newParseFailures),
// issue #285: record the judged boundary for the durable scan gate.
...(judged !== undefined ? { last_judged_msg: judged } : {}),
})
return {
tag: "save",
Expand Down
21 changes: 21 additions & 0 deletions packages/opencode/src/goal/loop.ts
Original file line number Diff line number Diff line change
Expand Up @@ -242,6 +242,26 @@ const serviceLayer = Layer.effect(
// newer revision). If this process already evaluated the CURRENT
// revision, the scan trigger is stale — skip.
if (scanResume && evaluatedRevisions.get(sessionID) === (goalState.revision ?? 0)) return
// issue #285 — durable boundary gate (scan path only). The
// evaluatedRevisions map above is process-local and dies with the
// process; the goal row's last_judged_msg is the crash-surviving record
// of which boundary was already judged and committed. While the session
// window still ends on that same message, no new progress has landed —
// re-judging would inflate turns_used and dispatch a duplicate
// continuation. Live idle events are never gated here: every dispatched
// continuation produces a fresh assistant message, so the live path
// always judges a new boundary.
if (scanResume && goalState.last_judged_msg) {
const win = yield* sessions
.messages({ sessionID, limit: 20 })
.pipe(
Effect.catchIf((e) => NotFoundError.isInstance(e), () =>
Effect.succeed([] as SessionV1.WithParts[]),
),
)
const lastSeen = [...win].reverse().find((m) => m.info.role === "assistant")
if (lastSeen && lastSeen.info.id === goalState.last_judged_msg) return
}
const goalOwner = { kind: "goal" as const, id: goalState.goal_id ?? "legacy" }
yield* automation.register(sessionID, goalOwner)
const observedLease = Option.getOrUndefined(yield* automation.claim(sessionID, goalOwner))
Expand Down Expand Up @@ -363,6 +383,7 @@ const serviceLayer = Layer.effect(
goalID: goalState.goal_id ?? "legacy",
revision: goalState.revision ?? 0,
},
lastAssistant.info.id,
),
),
)
Expand Down
6 changes: 6 additions & 0 deletions packages/opencode/src/goal/state.ts
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,11 @@ export class Info extends Schema.Class<Info>("GoalState")({
last_verdict: Schema.optional(Verdict),
last_reason: Schema.optional(Schema.String),
paused_reason: Schema.optional(Schema.String),
// issue #285 — the assistant message ID of the boundary judged by the last
// committed continue evaluation. Durable crash-recovery gate: the boot scan
// must not re-judge a window that still ends on this message (the
// process-local evaluatedRevisions map dies with the process).
last_judged_msg: Schema.optional(Schema.String),
consecutive_parse_failures: NonNegativeInt,
subgoals: Schema.Array(Schema.String).pipe(Schema.optional, Schema.withDecodingDefault(Effect.succeed([] as ReadonlyArray<string>))),
}) {}
Expand All @@ -54,6 +59,7 @@ export function advance(state: Info, patch: Partial<Omit<Info, "revision">>) {
last_verdict: state.last_verdict,
last_reason: state.last_reason,
paused_reason: state.paused_reason,
last_judged_msg: state.last_judged_msg,
consecutive_parse_failures: state.consecutive_parse_failures,
subgoals: state.subgoals,
...patch,
Expand Down
145 changes: 143 additions & 2 deletions packages/opencode/test/goal/e2e-loop.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -51,9 +51,9 @@ const assistantText = "I have made progress on the feature."
// Each scenario yields one scheduler turn before its first idle publish so that
// fiber can acquire the PubSub subscription. No business event exists yet, so
// outcome completion remains separately observed through public state/events.
const mkAssistant = () =>
const mkAssistant = (id?: string) =>
({
info: { role: "assistant", time: { created: Date.now() } },
info: { id, role: "assistant", time: { created: Date.now() } },
parts: [{ type: "text", text: assistantText }],
}) as never

Expand Down Expand Up @@ -1793,3 +1793,144 @@ describe("GoalLoop — judge-chain defect degrades into the parse budget (GOAL-F
}),
)
})

// issue #285 / GOAL-FP-01-21: the boot scan must not re-judge a boundary the
// crashed process already judged and committed. The process-local
// evaluatedRevisions map dies with the process, so the DURABLE gate is the
// goal row's last_judged_msg: updateAfterJudge records the judged assistant
// message ID on every continue commit, and the scan path skips evaluation
// while the session window still ends on that same message. Live idle events
// are never gated (each dispatched continuation produces a fresh assistant
// message, so the live path always sees a new boundary).
describe("GoalLoop — boot scan must not re-evaluate an already-judged boundary (issue #285)", () => {
let judgeCalls = 0
let continuationCalls = 0
let boundaryID = "msg_boundary_a"
const reset = () => {
judgeCalls = 0
continuationCalls = 0
boundaryID = "msg_boundary_a"
}

const sessionMock = Layer.mock(Session.Service, {
messages: () => Effect.succeed([mkAssistant(boundaryID)]),
})
const promptMock = Layer.mock(SessionPrompt.Service, withIdleAdmission({
prompt: () => Effect.die("the direct prompt path is not exercised in this scenario"),
promptIfIdle: () =>
Effect.sync(() => {
continuationCalls += 1
return Option.none()
}),
}))
const judgeMock = Layer.succeed(
GoalLoopJudgeLLM,
GoalLoopJudgeLLM.of({
call: () =>
Effect.sync(() => {
judgeCalls += 1
return JSON.stringify({ verdict: "continue", reason: "more work" })
}),
}),
)

const boundaryLayer = GoalLoop.layer.pipe(
Layer.provide(sessionMock),
Layer.provide(promptMock),
Layer.provide(Layer.mock(Provider.Service, {})),
Layer.provide(judgeMock),
Layer.provideMerge(Goal.defaultLayer),
Layer.provideMerge(SessionStatus.defaultLayer),
Layer.provideMerge(EventV2Bridge.defaultLayer),
Layer.provideMerge(SessionAutomationLease.defaultLayer),
Layer.provideMerge(Database.defaultLayer),
)
const it = testEffect(boundaryLayer)

const seedSessionRow = (sessionID: SessionID, directory: string) =>
Effect.gen(function* () {
const { db } = yield* Database.Service
const projectID = ProjectSchema.ID.make(Bun.randomUUIDv7())
yield* db.insert(ProjectTable).values({
id: projectID,
worktree: AbsolutePath.make(directory),
sandboxes: [AbsolutePath.make(directory)],
})
yield* db.insert(SessionTable).values({
id: sessionID,
project_id: projectID,
slug: "test-session",
directory,
title: "test session",
version: "1",
time_created: Date.now(),
time_updated: Date.now(),
})
})

// Commits one continue evaluation ahead of the (re)boot — models a process
// that crashed right after the commit, before the continuation produced an
// assistant message.
const commitPriorBoundary = (sid: SessionID) =>
Effect.gen(function* () {
const goal = yield* Goal.Service
const before = yield* goal.load(sid)
const result = yield* goal.updateAfterJudge(
sid,
"continue",
"pre-crash commit",
false,
{ goalID: before!.goal_id ?? "legacy", revision: before!.revision ?? 0 },
boundaryID,
)
expect(result?.state.turns_used).toBe(1)
return result?.state
})

it.instance("scan with an unchanged boundary skips re-evaluation (no inflation)", () =>
Effect.gen(function* () {
reset()
const loop = yield* GoalLoop.Service
const goal = yield* Goal.Service
const directory = (yield* TestInstance).directory
const sid = SessionID.descending()
yield* seedSessionRow(sid, directory)
yield* goal.set(sid, "boundary guard", 5)
yield* commitPriorBoundary(sid)

yield* loop.init()
// Negative assertion: the scan runs in a forked fiber with no readiness
// signal on the skip path, so a bounded wait stands in for polling.
yield* Effect.sleep("300 millis")
expect(judgeCalls).toBe(0)
expect(continuationCalls).toBe(0)
const g = yield* goal.load(sid)
expect(g?.turns_used).toBe(1)
}),
)

it.instance("scan proceeds once the window advanced past the boundary", () =>
Effect.gen(function* () {
reset()
const loop = yield* GoalLoop.Service
const goal = yield* Goal.Service
const directory = (yield* TestInstance).directory
const sid = SessionID.descending()
yield* seedSessionRow(sid, directory)
yield* goal.set(sid, "boundary guard", 5)
yield* commitPriorBoundary(sid)
// The continuation finished and produced a NEW assistant message.
boundaryID = "msg_boundary_b"

yield* loop.init()
yield* pollWithTimeout(
Effect.sync(() => (judgeCalls >= 1 ? true : undefined)),
"scan never re-evaluated the advanced boundary",
"5 seconds",
)
const g = yield* goal.load(sid)
expect(g?.turns_used).toBe(2)
expect(continuationCalls).toBe(1)
}),
)
})
Loading