diff --git a/packages/core/src/background-monitor.ts b/packages/core/src/background-monitor.ts new file mode 100644 index 000000000000..16c6a4428e16 --- /dev/null +++ b/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() +const _sessionPids = new Map>() + +/** 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) +} diff --git a/packages/opencode/src/cli/cmd/run.ts b/packages/opencode/src/cli/cmd/run.ts index fad09c3a7add..e08bfda16b01 100644 --- a/packages/opencode/src/cli/cmd/run.ts +++ b/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) { diff --git a/packages/opencode/src/effect/runtime-flags.ts b/packages/opencode/src/effect/runtime-flags.ts index 58dc50d0278c..cc25edfafa96 100644 --- a/packages/opencode/src/effect/runtime-flags.ts +++ b/packages/opencode/src/effect/runtime-flags.ts @@ -52,6 +52,7 @@ export class Service extends ConfigService.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")), }) {} diff --git a/packages/opencode/src/tool/background-shell.ts b/packages/opencode/src/tool/background-shell.ts new file mode 100644 index 000000000000..60299a25c219 --- /dev/null +++ b/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 +}): Effect.Effect { + 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)}`, + }), + ) + }), + ) +} diff --git a/packages/opencode/src/tool/monitor.ts b/packages/opencode/src/tool/monitor.ts new file mode 100644 index 000000000000..d32378a49741 --- /dev/null +++ b/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 +} + +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, + 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, ctx: Tool.Context) => + run(params, ctx).pipe(Effect.orDie), + } + }), +) diff --git a/packages/opencode/src/tool/monitor.txt b/packages/opencode/src/tool/monitor.txt new file mode 100644 index 000000000000..125545a6495f --- /dev/null +++ b/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 diff --git a/packages/opencode/src/tool/registry.ts b/packages/opencode/src/tool/registry.ts index 541d1f4bbbd0..b94b05d463bc 100644 --- a/packages/opencode/src/tool/registry.ts +++ b/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, diff --git a/packages/opencode/test/tool/monitor.test.ts b/packages/opencode/test/tool/monitor.test.ts new file mode 100644 index 000000000000..ab75b0fbaf37 --- /dev/null +++ b/packages/opencode/test/tool/monitor.test.ts @@ -0,0 +1,291 @@ +import { afterEach, describe, expect, test } from "bun:test" +import { Database } from "@opencode-ai/core/database/database" +import { Effect, Layer } from "effect" +import { Agent } from "@/agent/agent" +import { BackgroundJob } from "@/background/job" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { EventV2Bridge } from "@/event-v2-bridge" +import { Config } from "@/config/config" +import { Ripgrep } from "@opencode-ai/core/ripgrep" +import { RuntimeFlags } from "@/effect/runtime-flags" +import { Session } from "@/session/session" +import { SessionRunState } from "@/session/run-state" +import { SessionStatus } from "@/session/status" +import { ToolRegistry } from "@/tool/registry" +import { Truncate } from "@/tool/truncate" +import { MonitorTool } from "../../src/tool/monitor" +import { testEffect } from "../lib/effect" +import { MessageID, SessionID } from "../../src/session/schema" +import { disposeAllInstances } from "../fixture/fixture" + +afterEach(async () => { + await disposeAllInstances() +}) + +const layer = Layer.mergeAll( + Agent.defaultLayer, + BackgroundJob.defaultLayer, + EventV2Bridge.defaultLayer, + Config.defaultLayer, + CrossSpawnSpawner.defaultLayer, + Session.defaultLayer, + SessionRunState.defaultLayer, + SessionStatus.defaultLayer, + Truncate.defaultLayer, + ToolRegistry.defaultLayer, + Database.defaultLayer, + RuntimeFlags.layer({ experimentalMonitor: true }), +).pipe(Layer.provide(Ripgrep.defaultLayer)) + +const it = testEffect(layer) + +const ref = { + providerID: "test" as any, + modelID: "test-model" as any, +} + +const seed = Effect.gen(function* () { + const session = yield* Session.Service + const chat = yield* session.create({ title: "MonitorTest" }) + const user = yield* session.updateMessage({ + id: MessageID.ascending(), + role: "user", + sessionID: chat.id, + agent: "build", + model: ref, + time: { created: Date.now() }, + }) + const assistant = { + id: MessageID.ascending(), + role: "assistant" as const, + parentID: user.id, + sessionID: chat.id, + mode: "build", + agent: "build", + cost: 0, + path: { cwd: chat.directory, root: chat.directory }, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + modelID: ref.modelID, + providerID: ref.providerID, + time: { created: Date.now() }, + } + yield* session.updateMessage(assistant) + return { chat, assistant } +}) + +const runMonitor = Effect.gen(function* () { + const info = yield* MonitorTool + const tool = yield* info.init() + return tool +}) + +describe("MonitorTool", () => { + it.instance("arms a monitor and returns immediately", () => + Effect.gen(function* () { + const { chat, assistant } = yield* seed + const monitor = yield* runMonitor + const promptCalls: Array<{ text: string }> = [] + + const ops = { + prompt: (input: any) => + Effect.sync(() => { + promptCalls.push(input.parts[0]) + }), + } + + const result = yield* monitor.execute( + { + command: "echo 'hello'", + description: "test monitor", + }, + { + sessionID: chat.id, + messageID: assistant.id, + agent: "build", + abort: new AbortController().signal, + extra: { promptOps: ops }, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + expect(result.output).toContain("Monitor armed") + expect(result.metadata.monitor).toBe(true) + expect(result.metadata.description).toBe("test monitor") + + // Wait briefly for the monitor to exit and trigger callbacks + yield* Effect.sleep("500 millis") + + expect(promptCalls.length).toBeGreaterThan(0) + expect(promptCalls[promptCalls.length - 1].text).toContain("Monitor exited") + }), + ) +}) + +if (process.env.OPENCODE_LIVE_MONITOR_TEST) { + describe("MonitorTool live", () => { + it.instance("watches a live command and emits events", () => + Effect.gen(function* () { + const { chat, assistant } = yield* seed + const monitor = yield* runMonitor + const promptCalls: Array<{ text: string }> = [] + + const ops = { + prompt: (input: any) => + Effect.sync(() => { + promptCalls.push(input.parts[0]) + }), + } + + const result = yield* monitor.execute( + { + command: "bash -c 'echo event1; sleep 0.2; echo event2; sleep 0.2'", + description: "live test", + }, + { + sessionID: chat.id, + messageID: assistant.id, + agent: "build", + abort: new AbortController().signal, + extra: { promptOps: ops }, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + expect(result.output).toContain("Monitor armed") + + yield* Effect.sleep("1 second") + + const events = promptCalls.filter((p) => p.text.includes("Event:")) + expect(events.length).toBeGreaterThanOrEqual(1) + }), + ) + + it.instance("watches file ./f for changes and emits events", () => + Effect.gen(function* () { + const { chat, assistant } = yield* seed + const monitor = yield* runMonitor + const promptCalls: Array<{ text: string }> = [] + + const ops = { + prompt: (input: any) => + Effect.sync(() => { + promptCalls.push(input.parts[0]) + }), + } + + // Pre-create the file so monitor finds it immediately + // Note: must include newline so monitor's line-splitter picks it up + const watchFile = `${chat.directory}/f` + yield* Effect.promise(() => Bun.write(watchFile, "change1\n")) + + // Monitor: first print cwd to verify directory, then poll for file + const result = yield* monitor.execute( + { + command: "bash -c 'while true; do if [ -f ./f ]; then cat ./f; rm ./f; fi; sleep 0.1; done'", + description: "watch file ./f for changes", + }, + { + sessionID: chat.id, + messageID: assistant.id, + agent: "build", + abort: new AbortController().signal, + extra: { promptOps: ops }, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + expect(result.output).toContain("Monitor armed") + expect(result.metadata.description).toBe("watch file ./f for changes") + + // Wait for monitor to detect pre-existing file + yield* Effect.sleep("500 millis") + + // Trigger second change + yield* Effect.promise(() => Bun.write(watchFile, "change2\n")) + yield* Effect.sleep("500 millis") + + // Trigger third change + yield* Effect.promise(() => Bun.write(watchFile, "change3\n")) + yield* Effect.sleep("500 millis") + + // Collect events + const events = promptCalls.filter((p) => p.text.includes("Event:")) + expect(events.length).toBeGreaterThanOrEqual(1) + + // Verify event contents + const texts = events.map((e) => e.text) + expect(texts.some((t) => t.includes("change1"))).toBe(true) + expect(texts.some((t) => t.includes("change2"))).toBe(true) + expect(texts.some((t) => t.includes("change3"))).toBe(true) + }), + ) + + it.instance("E2E: prompts for watch and then changes file", () => + Effect.gen(function* () { + const { chat, assistant } = yield* seed + const monitor = yield* runMonitor + const promptCalls: Array<{ text: string }> = [] + + const ops = { + prompt: (input: any) => + Effect.sync(() => { + promptCalls.push(input.parts[0]) + }), + } + + const watchFile = `${chat.directory}/watchme.txt` + + // Create initial file + yield* Effect.promise(() => Bun.write(watchFile, "initial\n")) + + // Simulate user prompt: "watch watchme.txt for changes" + // Use a polling loop that detects file changes and reads them + const result = yield* monitor.execute( + { + command: "bash -c 'while true; do if [ -f ./watchme.txt ]; then cat ./watchme.txt; rm ./watchme.txt; fi; sleep 0.1; done'", + description: "watch watchme.txt for changes", + }, + { + sessionID: chat.id, + messageID: assistant.id, + agent: "build", + abort: new AbortController().signal, + extra: { promptOps: ops }, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + expect(result.output).toContain("Monitor armed") + expect(result.metadata.description).toBe("watch watchme.txt for changes") + + // Give monitor time to detect initial file + yield* Effect.sleep("500 millis") + + // Simulate file change: user writes a line + yield* Effect.promise(() => Bun.write(watchFile, "first line\n")) + yield* Effect.sleep("500 millis") + + // Simulate another file change + yield* Effect.promise(() => Bun.write(watchFile, "second line\n")) + yield* Effect.sleep("500 millis") + + // Collect events + const events = promptCalls.filter((p) => p.text.includes("Event:")) + expect(events.length).toBeGreaterThanOrEqual(1) + + // Verify event contents + const texts = events.map((e) => e.text) + expect(texts.some((t) => t.includes("first line"))).toBe(true) + expect(texts.some((t) => t.includes("second line"))).toBe(true) + }), + ) + }) +}