Merge pull request #283 from LeXwDeX/fix/goal-visibility

fix(goal): tolerate vanished-session message windows, surface loop failures

Author: LeXwDeX

packages/opencode/src/goal/judge.ts

@@ -75,10 +75,18 @@ export const run = Effect.fn("Goal.Judge.run")(function* (
     // "judge is unreliable" uniformly regardless of failure mode. The verdict
     // stays "continue" so a single transient blip does not stall the loop;
     // it only pauses after MAX_CONSECUTIVE_PARSE_FAILURES in a row.
-    Effect.orElseSucceed((): JudgeResult => ({
-      verdict: "continue",
-      reason: "judge transport error (timeout or network) — counting toward pause budget",
-      parseFailed: true,
-    })),
+    //
+    // catchCause (not orElseSucceed): the production callLLM chain can
+    // DEFECT — config first-use orDie, payload decode throws — and a defect
+    // escaping here kills afterIdle invisibly (the loop stalls at 0 turns
+    // with zero logs and no pause budget). catchCause folds defects into
+    // the same parseFailed budget.
+    Effect.catchCause(() =>
+      Effect.succeed({
+        verdict: "continue",
+        reason: "judge transport error (timeout or network) — counting toward pause budget",
+        parseFailed: true,
+      } satisfies JudgeResult),
+    ),
   )
 })

packages/opencode/src/goal/loop.ts

@@ -1,6 +1,7 @@
 export * as GoalLoop from "./loop"
 
 import { Effect, Layer, Context, Option, Stream, Scope, Fiber, Cause } from "effect"
+import { SessionV1 } from "@opencode-ai/core/v1/session"
 import { LayerNode } from "@opencode-ai/core/effect/layer-node"
 import { InstanceState } from "@/effect/instance-state"
 import { EventV2Bridge } from "@/event-v2-bridge"
@@ -14,6 +15,7 @@ import { GoalPrompts } from "./prompts"
 import { generateText } from "ai"
 import { SessionID } from "@/session/schema"
 import { SessionAutomationLease } from "@/session/automation-lease"
+import { NotFoundError } from "@/storage/storage"
 
 export interface Interface {
   readonly init: () => Effect.Effect<void>
@@ -148,12 +150,12 @@ const serviceLayer = Layer.effect(
               yield* triggerEvaluation(sid)
               // P2-B subscription survival: this handler now contains the
               // first defect-capable durable reads in the goal idle path
-              // (Goal.ownsSession / goal.load both orDie). Effect.ignore does
-              // NOT absorb defects — a transient store failure would
-              // permanently kill the runForEach subscription and the loop
-              // would never evaluate another idle event. catchCause absorbs
-              // failures AND defects at the boundary, so a store defect
-              // degrades to a logged, skipped evaluation — never a dead loop.
+              // (Goal.ownsSession / goal.load both orDie). In effect v4,
+              // Effect.ignore absorbs failures, defects AND interruptions —
+              // an error here would vanish without a trace, leaving skipped
+              // evaluations permanently invisible. catchCause keeps the same
+              // absorption but LOGS at the boundary, so a store defect
+              // degrades to a logged, skipped evaluation — never a silent one.
             }).pipe(
               Effect.catchCause((cause) =>
                 Effect.logWarning("GoalLoop idle handler failed", { sessionID: evt.data.sessionID, cause }),
@@ -263,7 +265,13 @@ const serviceLayer = Layer.effect(
         goalState.turns_used === 0 &&
         Date.now() - goalState.created_at > GoalPrompts.FRESHNESS_THRESHOLD
       ) {
-        const probeMsgs = yield* sessions.messages({ sessionID, limit: 1 })
+        const probeMsgs = yield* sessions
+          .messages({ sessionID, limit: 1 })
+          .pipe(
+            Effect.catchIf((e) => NotFoundError.isInstance(e), () =>
+              Effect.succeed([] as SessionV1.WithParts[]),
+            ),
+          )
         const hasAssistant = probeMsgs.some((m) => m.info.role === "assistant")
         if (isStaleZombie(goalState, hasAssistant)) {
           yield* pauseGoal(
@@ -274,7 +282,18 @@ const serviceLayer = Layer.effect(
         }
       }
 
-      const msgs = yield* sessions.messages({ sessionID, limit: 20 })
+      // A session whose row is gone (deleted mid-goal, or a synthetic
+      // session) fails page() with NotFoundError. Treat it as an empty window
+      // (same pattern as MessageV2.stream) so the no-lastAssistant branch
+      // below pauses visibly instead of this typed failure escaping and
+      // leaving the goal permanently "active".
+      const msgs = yield* sessions
+        .messages({ sessionID, limit: 20 })
+        .pipe(
+          Effect.catchIf((e) => NotFoundError.isInstance(e), () =>
+            Effect.succeed([] as SessionV1.WithParts[]),
+          ),
+        )
       const lastAssistant = [...msgs].reverse().find((m) => m.info.role === "assistant")
       if (!lastAssistant) {
         // No assistant message in the last 20 — the conversation may have
@@ -391,7 +410,14 @@ const serviceLayer = Layer.effect(
             sessionID,
             noReply: true,
             parts: [{ type: "text", text: updateResult.message }],
-          }).pipe(Effect.ignore)
+          }).pipe(
+            Effect.catchCause((cause) =>
+              Effect.logWarning("goal pause message delivery failed", {
+                sessionID,
+                cause: Cause.pretty(cause),
+              }),
+            ),
+          )
         }
         return
       }
@@ -409,8 +435,17 @@ const serviceLayer = Layer.effect(
       }
 
       // Reload messages after judge LLM call — the snapshot from before judge
-      // may be stale if user sent messages during the 5-30s judge latency
-      const freshMsgs = yield* sessions.messages({ sessionID, limit: 20 })
+      // may be stale if user sent messages during the 5-30s judge latency.
+      // Same vanished-session tolerance as the pre-judge window: NotFoundError
+      // becomes an empty window (shouldPreempt is defensively false for it),
+      // never a typed failure escaping the fork.
+      const freshMsgs = yield* sessions
+        .messages({ sessionID, limit: 20 })
+        .pipe(
+          Effect.catchIf((e) => NotFoundError.isInstance(e), () =>
+            Effect.succeed([] as SessionV1.WithParts[]),
+          ),
+        )
 
       if (shouldPreempt(freshMsgs)) {
         // Same self-interrupt hazard as the done branch above: we ARE the
@@ -561,7 +596,20 @@ const serviceLayer = Layer.effect(
       // afterIdle re-checks at its own entry load (see there) to close the
       // window between this load and the fork.
       if (scanResume && evaluatedRevisions.get(sessionID) === (goalState.revision ?? 0)) return
-      const fiber = yield* afterIdle(sessionID, scanResume).pipe(Effect.ignore, Effect.forkIn(scope))
+      // GOAL-FP-01-17: never Effect.ignore here. A typed failure escaping
+      // afterIdle (e.g. a messages read against a vanished session row) used
+      // to vanish into ignore and left the goal permanently "active" with
+      // zero logs — an invisible stall. Interrupts (fiber replacement by a
+      // newer idle, scope disposal) stay silent: they are the normal
+      // overwrite path, same F1 discipline as the continuation catch below.
+      const fiber = yield* afterIdle(sessionID, scanResume).pipe(
+        Effect.catchCause((cause) =>
+          Cause.hasInterrupts(cause)
+            ? Effect.void
+            : Effect.logWarning("goal afterIdle failed", { sessionID, cause: Cause.pretty(cause) }),
+        ),
+        Effect.forkIn(scope),
+      )
       yield* goal.registerLoopFiber(sessionID, fiber)
       yield* Fiber.await(fiber).pipe(
         Effect.flatMap(() => goal.clearLoopFiberIf(sessionID, fiber)),

packages/opencode/test/cli/github-action.test.ts

@@ -1,7 +1,6 @@
 import { test, expect, describe } from "bun:test"
 import { SessionV1 } from "@opencode-ai/core/v1/session"
 import { extractResponseText, formatPromptTooLargeError } from "../../src/cli/cmd/github"
-import type { MessageV2 } from "../../src/session/message-v2"
 import { SessionID, MessageID, PartID } from "../../src/session/schema"
 
 // Helper to create minimal valid parts

packages/opencode/test/dag/dag-loop-integration.test.ts

@@ -1,6 +1,5 @@
 import { describe, expect, it } from "bun:test"
 import {
-  buildGraph,
   type SchedulingNode,
   WorkflowRuntime,
 } from "@opencode-ai/core/dag/core/scheduling"

packages/opencode/test/goal/bootstrap-wiring.test.ts

@@ -0,0 +1,70 @@
+import { describe, expect } from "bun:test"
+import { Effect, Layer } from "effect"
+import { NodeFileSystem } from "@effect/platform-node"
+import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
+import { Goal } from "@/goal/goal"
+import { SessionStatus } from "@/session/status"
+import { SessionID } from "@/session/schema"
+import { InstanceStore } from "@/project/instance-store"
+import { provideTmpdirInstance } from "../fixture/fixture"
+import { pollWithTimeout, testEffect } from "../lib/effect"
+
+const it = testEffect(Layer.mergeAll(CrossSpawnSpawner.defaultLayer, NodeFileSystem.layer))
+
+// GOAL-BOOT-WIRING probe: boots the instance through the PRODUCTION path
+// (AppRuntime → InstanceStore.provide → InstanceBootstrap.run, which is the
+// only place that serviceOption-resolves and inits GoalLoop). Every other
+// test/goal suite builds GoalLoop.layer directly and therefore never
+// exercises this wiring. Pipeline under test, no judge involved:
+//   boot instance → set goal → publish session idle →
+//   afterIdle must reach the no-lastAssistant branch and PAUSE the goal.
+// If the serviceOption wiring, subscription, ownership gate, lease claim,
+// or event delivery is broken, the goal stays "active" and this test goes red.
+describe("GoalLoop production wiring — idle must drive afterIdle", () => {
+  it.live(
+    "an idle session with an active goal leaves the active state",
+    () =>
+      provideTmpdirInstance((path) =>
+        Effect.promise(async () => {
+          const { AppRuntime } = await import("@/effect/app-runtime")
+          await AppRuntime.runPromise(
+            Effect.gen(function* () {
+              const store = yield* InstanceStore.Service
+              yield* store.provide(
+                { directory: path },
+                Effect.gen(function* () {
+                  // Instance booted via production bootstrap — GoalLoop.init
+                  // must already have run here; do NOT call it again.
+                  const goal = yield* Goal.Service
+                  const status = yield* SessionStatus.Service
+
+                  const sid = SessionID.descending()
+                  yield* goal.set(sid, "wiring probe", 5)
+
+                  // Session starts busy; flip to idle — this is the exact event
+                  // the production Runner emits after a turn ends.
+                  yield* status.set(sid, { type: "busy" })
+                  yield* status.set(sid, { type: "idle" })
+
+                  // No assistant message exists → the healthy pipeline pauses
+                  // the goal with the "近期消息中无 assistant 回复" reason. A
+                  // stalled pipeline leaves it active.
+                  const final = yield* pollWithTimeout(
+                    Effect.gen(function* () {
+                      const state = yield* goal.load(sid)
+                      if (state && state.status !== "active") return state
+                      return undefined
+                    }),
+                    "goal never left active after idle — GoalLoop pipeline not armed on the production wiring",
+                    "8 seconds",
+                  )
+                  expect(final.status).toBe("paused")
+                }),
+              )
+            }),
+          )
+        }),
+      ),
+    20_000,
+  )
+})