feat(opencode): add Monitor tool (per-line background watcher)

packages/core/src/background-monitor.ts

@@ -0,0 +1,58 @@
+export * as BackgroundMonitor from "./background-monitor"
+
+// Process-local liveness tracking for the Monitor tool. The monitored process
+// itself runs as a BackgroundJob entry; this module only tracks "is a monitor
+// still running for this session" + its child PID, as PLAIN functions (no Effect
+// service) so the non-Effect CLI loop in cli/cmd/run.ts can read the count / kill
+// the PIDs SYNCHRONOUSLY — to defer process exit while a monitor is live and to
+// clean up on SIGINT.
+
+const _sessionCounts = new Map<string, number>()
+const _sessionPids = new Map<string, Array<number>>()
+
+/** Number of live monitor jobs for a session. */
+export function getMonitorCount(sessionID: string): number {
+  return _sessionCounts.get(sessionID) ?? 0
+}
+
+/** Bump the live count for a session (call when a job is armed). */
+export function monitorStarted(sessionID: string): void {
+  _sessionCounts.set(sessionID, (_sessionCounts.get(sessionID) ?? 0) + 1)
+}
+
+/** Drop the live count (and forget the PID) when a job ends. */
+export function monitorStopped(sessionID: string, pid?: number): void {
+  const next = (_sessionCounts.get(sessionID) ?? 0) - 1
+  if (next <= 0) _sessionCounts.delete(sessionID)
+  else _sessionCounts.set(sessionID, next)
+  if (pid !== undefined) removePid(sessionID, pid)
+}
+
+/** Track a child PID so SIGINT can kill it synchronously. */
+export function monitorPid(sessionID: string, pid: number): void {
+  const list = _sessionPids.get(sessionID) ?? []
+  list.push(pid)
+  _sessionPids.set(sessionID, list)
+}
+
+function removePid(sessionID: string, pid: number): void {
+  const list = _sessionPids.get(sessionID)
+  if (!list) return
+  const filtered = list.filter((p) => p !== pid)
+  if (filtered.length === 0) _sessionPids.delete(sessionID)
+  else _sessionPids.set(sessionID, filtered)
+}
+
+/** SIGTERM every tracked PID for a session (CLI SIGINT / shutdown path). */
+export function stopAllForSessionSync(sessionID: string): void {
+  const pids = _sessionPids.get(sessionID)
+  if (!pids) return
+  for (const pid of pids) {
+    try {
+      process.kill(pid, "SIGTERM")
+    } catch {
+      // already-exited or permission errors are fine
+    }
+  }
+  _sessionPids.delete(sessionID)
+}

packages/opencode/src/cli/cmd/run.ts

@@ -1,5 +1,6 @@
 import type { PermissionV1 } from "@opencode-ai/core/v1/permission"
 import { FSUtil } from "@opencode-ai/core/fs-util"
+import { getMonitorCount, stopAllForSessionSync } from "@opencode-ai/core/background-monitor"
 // CLI entry point for `opencode run` and `opencode --mini`.
 //
 // Handles three modes:
@@ -773,7 +774,8 @@ export const RunCommand = effectCmd({
               event.properties.sessionID === sessionID &&
               event.properties.status.type === "idle"
             ) {
-              break
+              if (getMonitorCount(sessionID) === 0) break
+              continue
             }
 
             if (event.type === "permission.asked") {
@@ -814,10 +816,20 @@ export const RunCommand = effectCmd({
             console.error(e)
             process.exitCode = 1
           })
+          process.on("SIGINT", () => {
+            stopAllForSessionSync(sessionID)
+            process.exit(1)
+          })
+
           async function finish() {
             if (args.attach) return
             const error = await completed
             if (error) process.exitCode = 1
+
+            // Keep the process alive while background monitors are active.
+            while (getMonitorCount(sessionID) > 0) {
+              await new Promise((resolve) => setTimeout(resolve, 100))
+            }
           }
 
           if (args.command) {

packages/opencode/src/effect/runtime-flags.ts

@@ -52,6 +52,7 @@ export class Service extends ConfigService.Service<Service>()("@opencode/Runtime
   bashDefaultTimeoutMs: positiveInteger("OPENCODE_EXPERIMENTAL_BASH_DEFAULT_TIMEOUT_MS"),
   experimentalNativeLlm: bool("OPENCODE_EXPERIMENTAL_NATIVE_LLM"),
   experimentalWebSockets: bool("OPENCODE_EXPERIMENTAL_WEBSOCKETS"),
+  experimentalMonitor: enabledByExperimental("OPENCODE_EXPERIMENTAL_MONITOR"),
   client: Config.string("OPENCODE_CLIENT").pipe(Config.withDefault("cli")),
 }) {}
 

packages/opencode/src/tool/background-shell.ts

@@ -0,0 +1,74 @@
+import { Cause, Effect, Stream } from "effect"
+import { ChildProcess } from "effect/unstable/process"
+import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner"
+import { Shell } from "@opencode-ai/core/shell"
+import { monitorPid, monitorStarted, monitorStopped } from "@opencode-ai/core/background-monitor"
+
+/** Build a detached login-shell command (same env/flags the old monitor used). */
+export function makeShellCommand(command: string, cwd: string): ChildProcess.Command {
+  const shell = Shell.acceptable()
+  const args = Shell.args(shell, command, cwd)
+  return ChildProcess.make(shell, args, {
+    cwd,
+    detached: process.platform !== "win32",
+    env: {
+      ...process.env,
+      TERM: "xterm-256color",
+      PAGER: "cat",
+      GIT_PAGER: "cat",
+    },
+  })
+}
+
+/**
+ * A fully-provided (`R = never`) Effect suitable as a `BackgroundJob` `run`:
+ * spawns `command`, tracks session liveness + PID (so the CLI defers exit and
+ * SIGINT can kill it), streams stdout line-by-line to `onLine` (when given),
+ * and resolves to the exit reason string. The process is killed when the job's
+ * scope closes (i.e. on `BackgroundJob.cancel` / session teardown) via the
+ * scoped spawn + an explicit kill finalizer.
+ */
+export function runShellJob(opts: {
+  sessionID: string
+  command: ChildProcess.Command
+  onLine?: (line: string) => Effect.Effect<void>
+}): Effect.Effect<string, unknown, ChildProcessSpawner> {
+  return Effect.scoped(
+    Effect.gen(function* () {
+      const spawner = yield* ChildProcessSpawner
+      monitorStarted(opts.sessionID)
+      let pid: number | undefined
+      yield* Effect.addFinalizer(() => Effect.sync(() => monitorStopped(opts.sessionID, pid)))
+
+      const handle = yield* spawner.spawn(opts.command)
+      pid = Number(handle.pid)
+      monitorPid(opts.sessionID, pid)
+      yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore))
+
+      if (opts.onLine) {
+        const buffer = { value: "" }
+        yield* Stream.runForEach(handle.stdout, (chunk) =>
+          Effect.gen(function* () {
+            buffer.value += new TextDecoder().decode(chunk as Uint8Array)
+            const lines = buffer.value.split(/\r?\n/)
+            buffer.value = lines.pop() ?? ""
+            for (const line of lines) {
+              const trimmed = line.trim()
+              if (!trimmed) continue
+              yield* opts.onLine!(trimmed)
+            }
+          }),
+        )
+        const trailing = buffer.value.trim()
+        if (trailing) yield* opts.onLine!(trailing)
+      }
+
+      return yield* handle.exitCode.pipe(
+        Effect.matchCause({
+          onSuccess: (code) => `exit code ${code}`,
+          onFailure: (cause) => `signal or error: ${Cause.squash(cause)}`,
+        }),
+      )
+    }),
+  )
+}

packages/opencode/src/tool/monitor.ts

@@ -0,0 +1,118 @@
+import * as Tool from "./tool"
+import DESCRIPTION from "./monitor.txt"
+import { BackgroundJob } from "@/background/job"
+import { Session } from "@/session/session"
+import { RuntimeFlags } from "@/effect/runtime-flags"
+import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner"
+import { Effect, Schema } from "effect"
+import { makeShellCommand, runShellJob } from "./background-shell"
+
+const id = "monitor"
+const TYPE = "monitor"
+
+export const Parameters = Schema.Struct({
+  command: Schema.String.annotate({
+    description: "The shell command to run. It must keep running and emit one stdout line per actual event.",
+  }),
+  description: Schema.String.annotate({
+    description: "A short description of what is being watched. Used in event messages.",
+  }),
+})
+
+type MonitorOps = {
+  prompt: (input: {
+    sessionID: string
+    agent: string
+    parts: Array<{ type: "text"; synthetic: boolean; text: string }>
+    noReply?: boolean
+  }) => Effect.Effect<void>
+}
+
+export const MonitorTool = Tool.define(
+  id,
+  Effect.gen(function* () {
+    const jobs = yield* BackgroundJob.Service
+    const sessions = yield* Session.Service
+    const flags = yield* RuntimeFlags.Service
+    const spawner = yield* ChildProcessSpawner
+
+    const run = Effect.fn("MonitorTool.execute")(function* (
+      params: Schema.Schema.Type<typeof Parameters>,
+      ctx: Tool.Context,
+    ) {
+      if (!flags.experimentalMonitor) {
+        return yield* Effect.die(new Error("Monitor tool requires OPENCODE_EXPERIMENTAL_MONITOR=true"))
+      }
+
+      const ops = (ctx.extra?.promptOps ?? undefined) as MonitorOps | undefined
+      if (!ops) {
+        return yield* Effect.die(new Error("Monitor tool requires promptOps in ctx.extra"))
+      }
+
+      yield* ctx.ask({
+        permission: id,
+        patterns: [params.command],
+        always: ["*"],
+        metadata: { description: params.description, command: params.command },
+      })
+
+      const session = yield* sessions.get(ctx.sessionID).pipe(Effect.orDie)
+
+      // Re-arming replaces the previous watch — cancel only prior MONITOR jobs
+      // for this session (leave any other background jobs running).
+      const existing = yield* jobs.list()
+      yield* Effect.forEach(
+        existing.filter((j) => j.type === TYPE && j.metadata?.["sessionId"] === ctx.sessionID),
+        (j) => jobs.cancel(j.id),
+        { concurrency: "unbounded", discard: true },
+      )
+
+      const command = makeShellCommand(params.command, session.directory)
+
+      // Each stdout line wakes the model; a clean process exit injects one final note.
+      const emit = (text: string) =>
+        ops
+          .prompt({
+            sessionID: ctx.sessionID,
+            agent: ctx.agent,
+            parts: [{ type: "text", synthetic: true, text }],
+          })
+          .pipe(Effect.ignore)
+
+      const job = runShellJob({
+        sessionID: ctx.sessionID,
+        command,
+        onLine: (line) => emit(`[Monitor: ${params.description}] Event: ${line}`),
+      }).pipe(
+        Effect.tap((reason) =>
+          emit(
+            `[Monitor: ${params.description}] Monitor exited (${reason}). If you still need to watch, re-arm with a working command.`,
+          ),
+        ),
+        Effect.provideService(ChildProcessSpawner, spawner),
+      )
+
+      const info = yield* jobs.start({
+        type: TYPE,
+        title: params.description,
+        // background:true => "born promoted": the tool returns immediately and
+        // nobody awaits the job inline. sessionId drives session-teardown cleanup.
+        metadata: { background: true, sessionId: ctx.sessionID, description: params.description },
+        run: job,
+      })
+
+      return {
+        title: params.description,
+        metadata: { monitor: true, monitorId: info.id, description: params.description },
+        output: `Monitor armed (${info.id}) for "${params.description}". Events will arrive as new messages. Do not re-arm it unless you need a different watch.`,
+      }
+    })
+
+    return {
+      description: DESCRIPTION,
+      parameters: Parameters,
+      execute: (params: Schema.Schema.Type<typeof Parameters>, ctx: Tool.Context) =>
+        run(params, ctx).pipe(Effect.orDie),
+    }
+  }),
+)

packages/opencode/src/tool/monitor.txt

@@ -0,0 +1,31 @@
+Run a long-lived command in the background and get notified whenever it emits a new line on stdout. The tool returns immediately after arming the monitor; each subsequent non-empty stdout line wakes the model with a new message. Use this for watching files, processes, or any event source that produces one line per event.
+
+IMPORTANT:
+- The command must keep running and emit ONE line per NEW event. It must not exit immediately.
+- Emit only on a real change. Never re-print content that has not changed — a loop that re-runs `tail -N`, `cat`, or `tail -5` re-emits the same lines every cycle, so once the watched job stops writing you get the same lines flooding back forever as bogus events. Prefer a streaming watcher that emits each new line exactly once.
+- Do not re-arm the same monitor repeatedly. If you need to change the watch, stop the old one first (or use a different description).
+- If the command exits on its own (e.g., the chosen watcher is not installed), you will get a wake-up message telling you it exited, so you can re-arm with a portable command.
+
+COVERAGE — silence is not success:
+- When watching a job or process for an outcome, your filter must match EVERY terminal state, not just the happy path. A monitor that greps only for the success marker stays silent through a crash, a hang, or an OOM kill — and silence looks exactly like "still running." Before arming, ask: if this process died right now, would my filter emit anything? If not, widen it.
+  - Wrong (silent on crash): `tail -F run.log | grep --line-buffered 'step='`
+  - Right: `tail -F run.log | grep -E --line-buffered 'step=|Traceback|Error|FAILED|Killed|OOM'`
+- If you cannot enumerate the failure signatures, broaden the grep alternation rather than narrow it — some extra noise beats missing a crashloop.
+
+STREAMS & BUFFERING:
+- Only stdout becomes events; stderr is not watched. For a command you run directly, merge stderr in so its failures reach your filter: `python train.py 2>&1 | grep -E --line-buffered ...`. (No effect when you `tail -F` an existing log — that file holds only what its writer redirected.)
+- Every pipe stage must flush per line or matches sit unseen in its buffer: `grep` needs `--line-buffered`, `awk` needs `fflush()`. Avoid `| head -N` — it cannot flush and delivers nothing until N matches accumulate.
+- In poll loops, swallow transient failures (`curl ... || true`) so one bad request doesn't kill the monitor.
+
+PATTERNS:
+- File changes: prefer `tail -F PATH` (follows forever, emits each new line once). If you must poll (e.g. a remote file over ssh), print only what changed, not a full re-dump — not `tail -5` / `cat` in a loop.
+- Directory changes: hash in a loop but emit ONLY when the hash changes: `while :; do h=$(find PATH -type f | sort | md5sum); [ "$h" != "$p" ] && echo "$h"; p=$h; sleep 2; done`. Do not echo the hash every iteration.
+- FINITE job (a build/run that will end): poll and EXIT the loop on a terminal state, so the monitor ends instead of polling a finished job forever:
+  `while :; do line=$(grep -E 'DONE|FAILED' PATH | tail -1); [ -n "$line" ] && { echo "$line"; break; }; sleep 5; done`
+  Do NOT rely on `tail -f PATH | grep -m1 DONE` to end cleanly: grep exits on the match, but tail only sees the closed pipe on its next write, so if the log goes quiet after the match the pipeline hangs until timeoutMs.
+- Poll intervals: 30s+ for remote APIs (rate limits), 0.5–1s for local checks.
+
+Parameters:
+- command: the shell command to run (must keep running and emit one line per event)
+- description: short description of what is being watched (used in event messages)
+- timeoutMs: optional maximum lifetime of the monitor in milliseconds; if omitted, it runs until the process exits or the session ends

packages/opencode/src/tool/registry.ts

@@ -29,6 +29,7 @@ import { WebSearchTool } from "./websearch"
 import { LspTool } from "./lsp"
 import * as Truncate from "./truncate"
 import { ApplyPatchTool } from "./apply_patch"
+import { MonitorTool } from "./monitor"
 import { Glob } from "@opencode-ai/core/util/glob"
 import path from "path"
 import { pathToFileURL } from "url"
@@ -96,6 +97,7 @@ export const layer = Layer.effect(
     const todo = yield* TodoWriteTool
     const lsptool = yield* LspTool
     const plan = yield* PlanExitTool
+    const monitor = yield* MonitorTool
     const webfetch = yield* WebFetchTool
     const websearch = yield* WebSearchTool
     const shell = yield* ShellTool
@@ -212,6 +214,7 @@ export const layer = Layer.effect(
           question: Tool.init(question),
           lsp: Tool.init(lsptool),
           plan: Tool.init(plan),
+          monitor: Tool.init(monitor),
         })
 
         return {
@@ -233,6 +236,7 @@ export const layer = Layer.effect(
             tool.patch,
             ...(flags.experimentalLspTool ? [tool.lsp] : []),
             ...(flags.experimentalPlanMode && flags.client === "cli" ? [tool.plan] : []),
+            ...(flags.experimentalMonitor ? [tool.monitor] : []),
           ],
           task: tool.task,
           read: tool.read,