From 187d0040337d60d7b91413af16845fcd6e727b7d Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Tue, 10 Mar 2026 08:18:28 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20watch=20tool=20=E2=80=94=20periodic=20c?= =?UTF-8?q?ommand=20polling=20within=20workstreams=20(#36)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: watch tool — periodic command polling within workstreams Add a new `watch` tool that lets the model (or user) set up periodic polling of a shell command. Results inject as synthetic user messages that trigger LLM turns, enabling reactive workflows like PR monitoring, CI/CD status tracking, and deployment health checks. Key design: - Single tool with create/list/cancel actions - Python expression DSL for stop conditions (restricted eval) - Server-owned WatchRunner daemon (DB-persisted, survives eviction + restart) - Three dispatch paths: idle, busy, and evicted workstream restore - REST API for console visibility (GET /v1/api/watches, POST cancel) - Migration 007, 8 storage CRUD methods, 75 new tests (1383 total) * fix: address Copilot review — condition errors, restore deadlock, docs - Condition eval errors now deactivate the watch immediately instead of silently looping until max_polls - Restored (evicted) workstreams set auto_approve=True to prevent approval deadlocks with no connected user - Tool description clarifies first-poll baseline behavior for change detection mode - Diagram updated: DELETE → POST /v1/api/watches/{id}/cancel --- docs/api-reference.md | 73 +++ docs/architecture.md | 1 + docs/diagrams/18-watch-architecture.puml | 166 ++++++ docs/diagrams/png/18-watch-architecture.png | 3 + docs/tools.md | 82 ++- tests/test_tools_schema.py | 3 +- tests/test_watch.py | 487 ++++++++++++++++++ tests/test_watch_storage.py | 130 +++++ turnstone/console/server.py | 88 ++++ turnstone/console/static/admin.js | 167 +++++- turnstone/console/static/index.html | 25 + turnstone/console/static/style.css | 14 + turnstone/core/auth.py | 7 + turnstone/core/session.py | 313 +++++++++++ turnstone/core/storage/_postgresql.py | 133 +++++ turnstone/core/storage/_protocol.py | 46 ++ turnstone/core/storage/_schema.py | 34 ++ turnstone/core/storage/_sqlite.py | 130 +++++ .../migrations/versions/007_watches.py | 47 ++ turnstone/core/watch.py | 442 ++++++++++++++++ turnstone/server.py | 122 ++++- turnstone/tools/watch.json | 36 ++ 22 files changed, 2543 insertions(+), 6 deletions(-) create mode 100644 docs/diagrams/18-watch-architecture.puml create mode 100644 docs/diagrams/png/18-watch-architecture.png create mode 100644 tests/test_watch.py create mode 100644 tests/test_watch_storage.py create mode 100644 turnstone/core/storage/migrations/versions/007_watches.py create mode 100644 turnstone/core/watch.py create mode 100644 turnstone/tools/watch.json diff --git a/docs/api-reference.md b/docs/api-reference.md index ef587e59..497a5f23 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -774,6 +774,79 @@ Status code: `400` --- +### `GET /v1/api/watches` + +List active watches on this server node. Optionally filter by workstream. +Requires `write` scope. + +**Query parameters:** + +| Parameter | Type | Required | Description | +|-----------|--------|----------|------------------------------------| +| `ws_id` | string | no | Filter to watches for this workstream. If omitted, returns all watches on the node. | + +**Response:** + +```json +{ + "watches": [ + { + "watch_id": "abc123def456...", + "ws_id": "ws-1", + "node_id": "host_a1b2", + "name": "pr-review", + "command": "gh pr view --json state", + "interval_secs": 300.0, + "stop_on": "data[\"state\"] == \"MERGED\"", + "max_polls": 100, + "poll_count": 5, + "last_output": "{\"state\": \"OPEN\"}", + "last_poll": "2026-03-09T12:00:00", + "next_poll": "2026-03-09T12:05:00", + "active": 1, + "created": "2026-03-09T11:30:00" + } + ] +} +``` + +--- + +### `POST /v1/api/watches/{watch_id}/cancel` + +Cancel an active watch. Sets `active=0` and clears `next_poll`. +Requires `write` scope. Verifies node ownership in multi-node deployments. + +**Path parameters:** + +| Parameter | Type | Description | +|------------|--------|-----------------| +| `watch_id` | string | Watch ID to cancel | + +**Response (success):** + +```json +{"status": "ok", "watch_id": "abc123def456..."} +``` + +**Error (not found):** + +```json +{"error": "Watch not found"} +``` + +Status code: `404` + +**Error (wrong node):** + +```json +{"error": "Watch belongs to another node"} +``` + +Status code: `403` + +--- + ### `OPTIONS` (any path) Handles CORS preflight requests. diff --git a/docs/architecture.md b/docs/architecture.md index 5cd88beb..649d6ded 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -44,6 +44,7 @@ turnstone/ tools.py Tool schema loader (JSON -> OpenAI function-calling format) mcp_client.py MCPClientManager — MCP server connections, tool discovery, dynamic refresh, async-sync bridge tool_search.py Dynamic tool search — BM25 index, session-scoped tool visibility + watch.py WatchRunner daemon — periodic command polling, condition DSL, result dispatch model_registry.py ModelRegistry — named model configs, lazy client creation, fallback routing memory.py Persistence facade (delegates to storage backend) storage/ Pluggable storage: StorageBackend protocol, SQLite + PostgreSQL diff --git a/docs/diagrams/18-watch-architecture.puml b/docs/diagrams/18-watch-architecture.puml new file mode 100644 index 00000000..d140030f --- /dev/null +++ b/docs/diagrams/18-watch-architecture.puml @@ -0,0 +1,166 @@ +@startuml +!theme plain +title Turnstone — Watch Tool Architecture + +skinparam participant { + BackgroundColor<> #FFE0B2 + BackgroundColor<> #B3E5FC + BackgroundColor<> #C8E6C9 + BackgroundColor<> #E8EAF6 +} + +participant "ChatSession\n(session.py)" as Session <> +participant "WatchRunner\n(watch.py)" as Runner <> +participant "StorageBackend\n(SQLite)" as Storage <> +participant "WebUI / SSE\n(server.py)" as UI <> + +== Create Phase == + +Session -> Session : _prepare_watch(action="create") +note right + Validates: + - command via is_command_blocked() + - poll_every → parse_duration() + - stop_on → validate_condition() + - max watches limit (5) + - duplicate name check + needs_approval = True +end note + +Session -> Storage : create_watch(watch_id, ws_id,\nnode_id, command, interval,\nstop_on, max_polls, next_poll) + +Session --> UI : tool_result:\n"Watch 'pr-review' created" + +== Poll Phase (WatchRunner daemon, every 15s) == + +Runner -> Storage : list_due_watches(now) +Storage --> Runner : due_watches[] +note right + Filters: + active=1 AND + next_poll <= now AND + node_id matches +end note + +loop for each due watch + + Runner -> Runner : is_command_blocked()? + alt blocked + Runner -> Storage : update_watch(active=False) + else safe + + Runner -> Runner : subprocess.run(command) + note right + timeout = tool_timeout + start_new_session = True + output truncated at 64KB + end note + + Runner -> Runner : evaluate_condition(\nstop_on, output,\nexit_code, prev_output) + note right + **Variables:** + output, data, exit_code, + prev_output, changed + + **Safe builtins only:** + len, str, int, sorted, ... + No import/open/exec/eval + + **stop_on=None:** + fires on change (skip 1st poll) + end note + + alt condition fired OR max_polls reached + Runner -> Storage : update_watch(\npoll_count++,\nlast_output, active=False) + Runner -> Runner : format_watch_message() + Runner -> Runner : _dispatch_result(ws_id, msg) + else not fired + Runner -> Storage : update_watch(\npoll_count++,\nlast_output, next_poll) + end + + end +end + +== Dispatch Phase == + +note over Runner, Session + **Three dispatch paths:** +end note + +alt Path A: workstream active + idle + Runner -> Session : dispatch_fn(message)\n→ _watch_pending.put() + Session -> Session : _dispatch_pending_watch()\n→ self.send(message) + Session -> UI : SSE: thinking, content,\ntool calls... + note right + Watch result appears as + synthetic user message. + Model sees it and responds. + Depth guard: max 5 chains. + end note + +else Path B: workstream active + busy + Runner -> Session : dispatch_fn(message)\n→ _watch_pending.put() + note right + Queued. Dispatched when + current send() reaches IDLE. + end note + +else Path C: workstream evicted + Runner -> Runner : restore_fn(ws_id) + note right + 1. mgr.create() — may evict + another idle workstream + 2. session.resume(ws_id) + 3. set_watch_runner() + 4. register new dispatch_fn + end note + Runner -> Session : restored dispatch_fn(message) +end + +== Cancel / List == + +Session -> Storage : list_watches_for_ws(ws_id) +note right : action="list" (auto-approve) + +Session -> Storage : update_watch(active=False) +note right : action="cancel" (auto-approve) + +== Server Lifecycle == + +note over Runner, Storage + **Startup:** + 1. WatchRunner created in main() with storage + node_id + 2. restore_fn closure captures WorkstreamManager + 3. Initial workstream: session.set_watch_runner(runner) + 4. _lifespan(): runner.start() — daemon thread begins + + **New workstream:** + session.set_watch_runner(runner) in create_workstream() + → registers dispatch_fn for ws_id + + **Eviction / close:** + session.close() → runner.remove_dispatch_fn(ws_id) + Watches remain active in DB — WatchRunner uses restore_fn + + **Restart recovery:** + Overdue watches fire ONE immediate poll + next_poll updated to now + interval + Normal cadence resumes + + **Shutdown:** + _lifespan(): runner.stop() — joins thread +end note + +== REST API == + +note over UI, Storage + **GET /v1/api/watches[?ws_id=X]** + List active watches (for node or workstream) + + **POST /v1/api/watches/{watch_id}/cancel** + Cancel a watch (sets active=False) + + Both require write scope +end note + +@enduml diff --git a/docs/diagrams/png/18-watch-architecture.png b/docs/diagrams/png/18-watch-architecture.png new file mode 100644 index 00000000..daedafae --- /dev/null +++ b/docs/diagrams/png/18-watch-architecture.png @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:96176a09e65e90dadc32d5e9ed778423842be89204d2cf382225f53a90cfaf01 +size 258547 diff --git a/docs/tools.md b/docs/tools.md index 59253f8d..d2fb20d4 100644 --- a/docs/tools.md +++ b/docs/tools.md @@ -1,6 +1,6 @@ # Tools Reference -turnstone exposes 15 built-in tools plus any number of external MCP tools to the +turnstone exposes 16 built-in tools plus any number of external MCP tools to the LLM via the OpenAI function-calling interface. Built-in tools are defined as JSON files under `turnstone/tools/` and loaded at startup by `turnstone/core/tools.py`. MCP tools are discovered from configured MCP servers at startup by @@ -46,12 +46,12 @@ schema plus turnstone-specific metadata keys: | Name | Description | |---------------------|-------------| -| `TOOLS` | All 15 tool definitions (sent to the model). | +| `TOOLS` | All 16 tool definitions (sent to the model). | | `AGENT_TOOLS` | Tools with `agent: true` -- available to plan sub-agents. Read-only tools. | | `TASK_AGENT_TOOLS` | Tools with `task_agent: true` -- available to task sub-agents. Includes write operations. | | `AGENT_AUTO_TOOLS` | Set of tool names with `auto_approve: true` -- no user confirmation needed. | | `TASK_AUTO_TOOLS` | Same as `AGENT_AUTO_TOOLS` (identical filter). | -| `BUILTIN_TOOL_NAMES`| Frozenset of all 15 built-in tool names. Used by tool search to distinguish always-on tools from deferrable MCP tools. | +| `BUILTIN_TOOL_NAMES`| Frozenset of all 16 built-in tool names. Used by tool search to distinguish always-on tools from deferrable MCP tools. | | `PRIMARY_KEY_MAP` | Dict mapping tool name to its `primary_key` parameter name. | --- @@ -422,6 +422,81 @@ Provide either `username` for user-based targeting or `channel_type` + --- +### watch + +Set up periodic polling of a shell command within the current workstream. +Results are injected back into the conversation as synthetic user messages, +triggering the model to respond and act. Use for monitoring CI/CD pipelines, +PR reviews, deployments, file changes, etc. + +| Parameter | Type | Required | Description | +|-------------|---------|----------|-------------| +| `action` | string | yes | `create`, `list`, or `cancel`. | +| `command` | string | create | Shell command to poll periodically. | +| `poll_every`| string | no | Poll interval as duration (`30s`, `5m`, `1h`). Default: `5m`. | +| `stop_on` | string | no | Python expression for stop condition (see below). Omit for change detection. | +| `name` | string | create | Human-readable watch name (e.g. `pr-review`). Used as identifier for cancel. | +| `max_polls` | integer | no | Max poll cycles before auto-cancel. Default: 100. | + +**Actions:** + +- `create` — Start a new watch. Requires approval (same as bash — runs shell + commands). Persists to the `watches` table; the server-level `WatchRunner` + daemon polls every 15 seconds for due watches. +- `list` — Show all active watches in this workstream. Auto-approved. +- `cancel` — Stop a watch by name or ID prefix. Auto-approved. + +**Stop condition DSL** — The `stop_on` parameter accepts a Python expression +evaluated after each poll. Available variables: + +| Variable | Type | Description | +|---------------|------------|-------------| +| `output` | `str` | stdout (+stderr) of the command. | +| `data` | `Any` | `json.loads(output)`, or `None` if not valid JSON. | +| `exit_code` | `int` | Process exit code. | +| `prev_output` | `str|None` | Previous poll's stdout (`None` on first poll). | +| `changed` | `bool` | `True` if output differs from previous poll. | + +Safe builtins: `len`, `str`, `int`, `float`, `bool`, `abs`, `min`, `max`, +`any`, `all`, `isinstance`, `sorted`. No `import`, `open`, `exec`, or +`eval`. Security model: equivalent to `bash` — the model already has shell +access. + +**Examples:** +``` +data["state"] == "MERGED" +"error" in output +exit_code != 0 +changed and "ready" in output.lower() +data.get("mergedAt") is not None +``` + +**Lifecycle:** + +1. Model calls `watch(action="create", ...)` — persisted to SQLite. +2. `WatchRunner` daemon polls for due watches every 15s. +3. Each poll runs the command, evaluates the condition. +4. When the condition fires (or max polls reached), the result is injected + as a synthetic user message and the watch auto-cancels. +5. If the workstream was evicted, it is restored before injection. +6. Watches survive server restart (overdue watches fire once on recovery). + +**Constraints:** + +- Max 5 active watches per workstream. +- Poll interval: 10s–24h. +- Output truncated at 64 KB. +- Max 5 consecutive watch dispatches per worker thread (depth guard). +- Duplicate names rejected within the same workstream. + +- **Auto-approve**: `create` requires approval; `list` and `cancel` are auto-approved. +- **Agent availability**: Main session only — not available to plan/task sub-agents. + +> See [Watch Architecture](diagrams/png/18-watch-architecture.png) for the +> full poll → evaluate → dispatch flow. + +--- + ## Summary Table | Tool | Category | Auto-approve | agent | task_agent | primary_key | @@ -441,6 +516,7 @@ Provide either `username` for user-based targeting or `channel_type` + | `recall` | Memory | Yes | No | No | `query` | | `forget` | Memory | Yes | No | No | `key` | | `notify` | Notify | Yes | Yes | Yes | `message` | +| `watch` | Monitor | No (create) | No | No | `command` | | `tool_search`| Search | Yes | No | No | `query` | --- diff --git a/tests/test_tools_schema.py b/tests/test_tools_schema.py index 02bf01d5..da8b3ffe 100644 --- a/tests/test_tools_schema.py +++ b/tests/test_tools_schema.py @@ -72,7 +72,7 @@ class TestToolsMetadata: """Validate the metadata extracted from JSON files.""" def test_tool_count(self): - assert len(TOOLS) == 15 + assert len(TOOLS) == 16 def test_agent_tools_count(self): assert len(AGENT_TOOLS) == 7 @@ -102,6 +102,7 @@ class TestToolsMetadata: "recall": "query", "forget": "key", "notify": "message", + "watch": "command", } assert expected == PRIMARY_KEY_MAP diff --git a/tests/test_watch.py b/tests/test_watch.py new file mode 100644 index 00000000..f2530566 --- /dev/null +++ b/tests/test_watch.py @@ -0,0 +1,487 @@ +"""Tests for the watch module — duration parsing, condition evaluation, WatchRunner.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from unittest.mock import MagicMock + +import pytest + +from turnstone.core.watch import ( + WatchRunner, + evaluate_condition, + format_interval, + format_watch_message, + parse_duration, + validate_condition, +) + +# --------------------------------------------------------------------------- +# parse_duration +# --------------------------------------------------------------------------- + + +class TestParseDuration: + def test_seconds(self): + assert parse_duration("30s") == 30.0 + + def test_minutes(self): + assert parse_duration("5m") == 300.0 + + def test_hours(self): + assert parse_duration("1h") == 3600.0 + + def test_compound(self): + assert parse_duration("2h30m") == 9000.0 + + def test_bare_number(self): + assert parse_duration("90") == 90.0 + + def test_bare_float(self): + assert parse_duration("10.5") == 10.5 + + def test_whitespace(self): + assert parse_duration(" 5m ") == 300.0 + + def test_case_insensitive(self): + assert parse_duration("1H30M") == 5400.0 + + def test_empty_raises(self): + with pytest.raises(ValueError, match="empty"): + parse_duration("") + + def test_invalid_raises(self): + with pytest.raises(ValueError, match="invalid duration"): + parse_duration("abc") + + def test_negative_raises(self): + with pytest.raises(ValueError, match="positive"): + parse_duration("-5") + + def test_zero_raises(self): + with pytest.raises(ValueError, match="positive"): + parse_duration("0") + + def test_zero_duration_raises(self): + with pytest.raises(ValueError, match="positive"): + parse_duration("0s") + + +# --------------------------------------------------------------------------- +# validate_condition +# --------------------------------------------------------------------------- + + +class TestValidateCondition: + def test_valid_expression(self): + assert validate_condition('data["state"] == "MERGED"') is None + + def test_valid_simple(self): + assert validate_condition('"error" in output') is None + + def test_valid_compound(self): + assert validate_condition('changed and "ready" in output.lower()') is None + + def test_syntax_error(self): + result = validate_condition("if True:") + assert result is not None + assert "syntax" in result.lower() + + def test_incomplete_expression(self): + result = validate_condition("==") + assert result is not None + + +# --------------------------------------------------------------------------- +# evaluate_condition +# --------------------------------------------------------------------------- + + +class TestEvaluateCondition: + def test_none_first_poll_no_fire(self): + """With stop_on=None, first poll (prev_output=None) should not fire.""" + fired, reason = evaluate_condition(None, "hello", 0, None) + assert not fired + + def test_none_change_detected(self): + fired, reason = evaluate_condition(None, "world", 0, "hello") + assert fired + assert "changed" in reason + + def test_none_no_change(self): + fired, reason = evaluate_condition(None, "same", 0, "same") + assert not fired + + def test_string_match(self): + fired, reason = evaluate_condition('"error" in output', "has error here", 0, None) + assert fired + + def test_string_no_match(self): + fired, reason = evaluate_condition('"error" in output', "all good", 0, None) + assert not fired + + def test_exit_code(self): + fired, reason = evaluate_condition("exit_code != 0", "fail", 1, None) + assert fired + + def test_exit_code_zero(self): + fired, reason = evaluate_condition("exit_code != 0", "ok", 0, None) + assert not fired + + def test_json_data(self): + output = '{"state": "MERGED"}' + fired, reason = evaluate_condition('data["state"] == "MERGED"', output, 0, None) + assert fired + + def test_json_data_no_match(self): + output = '{"state": "OPEN"}' + fired, reason = evaluate_condition('data["state"] == "MERGED"', output, 0, None) + assert not fired + + def test_json_data_none_for_non_json(self): + """Non-JSON output should have data=None.""" + fired, reason = evaluate_condition("data is None", "plain text", 0, None) + assert fired + + def test_changed_variable(self): + fired, reason = evaluate_condition("changed", "new", 0, "old") + assert fired + + def test_changed_false(self): + fired, reason = evaluate_condition("changed", "same", 0, "same") + assert not fired + + def test_compound_condition(self): + fired, reason = evaluate_condition( + 'changed and "ready" in output.lower()', + "System Ready", + 0, + "System Starting", + ) + assert fired + + def test_invalid_expression_no_crash(self): + fired, reason = evaluate_condition("1/0", "hello", 0, None) + assert not fired + assert "error" in reason.lower() + + def test_no_import_builtin(self): + """__import__ should not be accessible.""" + fired, reason = evaluate_condition("__import__('os')", "hello", 0, None) + assert not fired + assert "error" in reason.lower() + + def test_no_open_builtin(self): + fired, reason = evaluate_condition("open('/etc/passwd')", "hello", 0, None) + assert not fired + assert "error" in reason.lower() + + def test_no_exec_builtin(self): + fired, reason = evaluate_condition("exec('print(1)')", "hello", 0, None) + assert not fired + assert "error" in reason.lower() + + def test_no_eval_builtin(self): + fired, reason = evaluate_condition("eval('1+1')", "hello", 0, None) + assert not fired + assert "error" in reason.lower() + + def test_no_compile_builtin(self): + fired, reason = evaluate_condition("compile('1','','eval')", "hello", 0, None) + assert not fired + assert "error" in reason.lower() + + def test_safe_len(self): + fired, reason = evaluate_condition("len(output) > 0", "hello", 0, None) + assert fired + + def test_safe_sorted(self): + fired, reason = evaluate_condition("sorted([3,1,2]) == [1,2,3]", "x", 0, None) + assert fired + + def test_data_get_method(self): + output = '{"mergedAt": "2024-01-15"}' + fired, reason = evaluate_condition('data.get("mergedAt") is not None', output, 0, None) + assert fired + + def test_prev_output_available(self): + fired, reason = evaluate_condition( + "prev_output is not None and output != prev_output", + "new", + 0, + "old", + ) + assert fired + + +# --------------------------------------------------------------------------- +# format_interval +# --------------------------------------------------------------------------- + + +class TestFormatInterval: + def test_seconds(self): + assert format_interval(30) == "30s" + + def test_exactly_60(self): + assert format_interval(60) == "1m" + + def test_minutes(self): + assert format_interval(300) == "5m" + + def test_exactly_3600(self): + assert format_interval(3600) == "1h" + + def test_hours_and_minutes(self): + assert format_interval(5400) == "1h30m" + + def test_hours_only(self): + assert format_interval(7200) == "2h" + + def test_large_value(self): + assert format_interval(86400) == "24h" + + +# --------------------------------------------------------------------------- +# format_watch_message +# --------------------------------------------------------------------------- + + +class TestFormatWatchMessage: + def test_basic(self): + msg = format_watch_message( + name="pr-review", + command="gh pr view --json state", + output='{"state": "MERGED"}', + poll_count=5, + max_polls=100, + elapsed_secs=1500, + stop_on='data["state"] == "MERGED"', + is_final=True, + reason='condition met: data["state"] == "MERGED"', + ) + assert "pr-review" in msg + assert "poll #5/100" in msg + assert "25m" in msg + assert "gh pr view --json state" in msg + assert "MERGED" in msg + assert "auto-cancelled" in msg.lower() + # Model should see the condition it was waiting for + assert "condition:" in msg.lower() + + def test_non_final(self): + msg = format_watch_message( + name="deploy", + command="curl -s http://localhost/health", + output="ok", + poll_count=3, + max_polls=50, + elapsed_secs=90, + stop_on=None, + is_final=False, + reason="", + ) + assert "deploy" in msg + assert "auto-cancelled" not in msg.lower() + # Change-detection mode should be indicated + assert "output change" in msg.lower() + + def test_max_polls_final(self): + msg = format_watch_message( + name="test", + command="echo hello", + output="hello", + poll_count=100, + max_polls=100, + elapsed_secs=6000, + stop_on=None, + is_final=True, + reason="", + ) + assert "max polls" in msg.lower() + + +# --------------------------------------------------------------------------- +# WatchRunner +# --------------------------------------------------------------------------- + + +class TestWatchRunner: + def _make_runner(self, storage=None, **kwargs): + if storage is None: + storage = MagicMock() + storage.list_due_watches.return_value = [] + return WatchRunner( + storage=storage, + node_id="test-node", + check_interval=0.1, + tool_timeout=5, + **kwargs, + ) + + def test_start_stop(self): + runner = self._make_runner() + runner.start() + assert runner._thread is not None + assert runner._thread.is_alive() + runner.stop() + assert runner._thread is None + + def test_tick_calls_list_due(self): + storage = MagicMock() + storage.list_due_watches.return_value = [] + runner = self._make_runner(storage=storage) + runner._tick() + storage.list_due_watches.assert_called_once() + + def test_poll_watch_runs_command(self): + storage = MagicMock() + storage.update_watch.return_value = True + runner = self._make_runner(storage=storage) + dispatch_fn = MagicMock() + runner.set_dispatch_fn("ws-1", dispatch_fn) + + watch_row = { + "watch_id": "abc123", + "ws_id": "ws-1", + "name": "test-watch", + "command": "echo hello", + "stop_on": '"hello" in output', + "max_polls": 100, + "poll_count": 0, + "last_output": None, + "interval_secs": 60, + "created": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S"), + } + runner._poll_watch(watch_row) + + # Should update the watch in storage + storage.update_watch.assert_called_once() + call_kwargs = storage.update_watch.call_args + assert call_kwargs[0][0] == "abc123" # watch_id + assert call_kwargs[1]["poll_count"] == 1 + # Condition should fire (output contains "hello") + assert call_kwargs[1]["active"] is False # deactivated + # Should dispatch result + dispatch_fn.assert_called_once() + + def test_poll_watch_no_fire_on_first_change_detection(self): + storage = MagicMock() + storage.update_watch.return_value = True + runner = self._make_runner(storage=storage) + dispatch_fn = MagicMock() + runner.set_dispatch_fn("ws-1", dispatch_fn) + + watch_row = { + "watch_id": "abc123", + "ws_id": "ws-1", + "name": "test-watch", + "command": "echo hello", + "stop_on": None, # change detection + "max_polls": 100, + "poll_count": 0, + "last_output": None, # first poll + "interval_secs": 60, + "created": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S"), + } + runner._poll_watch(watch_row) + + # First poll with change detection should not fire + dispatch_fn.assert_not_called() + call_kwargs = storage.update_watch.call_args + # Watch should remain active + assert "active" not in call_kwargs[1] or call_kwargs[1].get("active") is not False + + def test_max_polls_deactivates(self): + storage = MagicMock() + storage.update_watch.return_value = True + runner = self._make_runner(storage=storage) + dispatch_fn = MagicMock() + runner.set_dispatch_fn("ws-1", dispatch_fn) + + watch_row = { + "watch_id": "abc123", + "ws_id": "ws-1", + "name": "test-watch", + "command": "echo hello", + "stop_on": '"never" in output', # won't fire + "max_polls": 5, + "poll_count": 4, # next is #5 = max + "last_output": "hello\n", + "interval_secs": 60, + "created": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S"), + } + runner._poll_watch(watch_row) + + call_kwargs = storage.update_watch.call_args + assert call_kwargs[1]["active"] is False + assert call_kwargs[1]["poll_count"] == 5 + dispatch_fn.assert_called_once() + + def test_blocked_command_deactivates(self): + storage = MagicMock() + storage.update_watch.return_value = True + runner = self._make_runner(storage=storage) + + watch_row = { + "watch_id": "abc123", + "ws_id": "ws-1", + "name": "test-watch", + "command": "rm -rf /", + "stop_on": None, + "max_polls": 100, + "poll_count": 0, + "last_output": None, + "interval_secs": 60, + "created": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S"), + } + runner._poll_watch(watch_row) + + storage.update_watch.assert_called_once() + call_kwargs = storage.update_watch.call_args + assert call_kwargs[0][0] == "abc123" + assert call_kwargs[1]["active"] is False + + def test_dispatch_fn_registry(self): + runner = self._make_runner() + fn1 = MagicMock() + fn2 = MagicMock() + + runner.set_dispatch_fn("ws-1", fn1) + runner.set_dispatch_fn("ws-2", fn2) + + runner._dispatch_result("ws-1", "msg1") + fn1.assert_called_once_with("msg1") + fn2.assert_not_called() + + runner.remove_dispatch_fn("ws-1") + # After removal, dispatch should try restore_fn + runner._dispatch_result("ws-1", "msg2") + fn1.assert_called_once() # still just the one call + + def test_restore_fn_called_for_evicted(self): + restored_fn = MagicMock() + restore_fn = MagicMock(return_value=restored_fn) + runner = self._make_runner(restore_fn=restore_fn) + + runner._dispatch_result("ws-evicted", "hello") + restore_fn.assert_called_once_with("ws-evicted") + restored_fn.assert_called_once_with("hello") + + def test_run_command_success(self): + runner = self._make_runner() + output, code = runner._run_command("echo hello") + assert "hello" in output + assert code == 0 + + def test_run_command_failure(self): + runner = self._make_runner() + output, code = runner._run_command("exit 42") + assert code == 42 + + def test_run_command_timeout(self): + runner = self._make_runner() + runner._tool_timeout = 1 + output, code = runner._run_command("sleep 30") + assert "timed out" in output.lower() + assert code == -1 diff --git a/tests/test_watch_storage.py b/tests/test_watch_storage.py new file mode 100644 index 00000000..f1b05a52 --- /dev/null +++ b/tests/test_watch_storage.py @@ -0,0 +1,130 @@ +"""Tests for watches storage CRUD.""" + +from __future__ import annotations + +import pytest + +from turnstone.core.storage._sqlite import SQLiteBackend + + +@pytest.fixture +def db(tmp_path): + """Fresh SQLite backend for each test.""" + return SQLiteBackend(str(tmp_path / "test.db")) + + +def _make_watch_kwargs(**overrides): + """Build default kwargs for create_watch.""" + defaults = { + "watch_id": "watch_001", + "ws_id": "ws-abc", + "node_id": "node-1", + "name": "pr-review", + "command": "gh pr view --json state", + "interval_secs": 300.0, + "stop_on": 'data["state"] == "MERGED"', + "max_polls": 100, + "created_by": "model", + "next_poll": "2099-01-01T00:05:00", + } + defaults.update(overrides) + return defaults + + +class TestWatchCRUD: + def test_create_and_get(self, db): + db.create_watch(**_make_watch_kwargs()) + w = db.get_watch("watch_001") + assert w is not None + assert w["name"] == "pr-review" + assert w["command"] == "gh pr view --json state" + assert w["interval_secs"] == 300.0 + assert w["active"] == 1 + assert w["poll_count"] == 0 + + def test_get_nonexistent(self, db): + assert db.get_watch("nope") is None + + def test_create_idempotent(self, db): + db.create_watch(**_make_watch_kwargs()) + db.create_watch(**_make_watch_kwargs()) # OR IGNORE + assert db.get_watch("watch_001") is not None + + def test_update(self, db): + db.create_watch(**_make_watch_kwargs()) + updated = db.update_watch( + "watch_001", + poll_count=5, + last_output="hello", + last_exit_code=0, + ) + assert updated is True + w = db.get_watch("watch_001") + assert w["poll_count"] == 5 + assert w["last_output"] == "hello" + assert w["last_exit_code"] == 0 + + def test_update_nonexistent(self, db): + assert db.update_watch("nope", poll_count=1) is False + + def test_update_active_flag(self, db): + db.create_watch(**_make_watch_kwargs()) + db.update_watch("watch_001", active=False) + w = db.get_watch("watch_001") + assert w["active"] == 0 + + def test_delete(self, db): + db.create_watch(**_make_watch_kwargs()) + assert db.delete_watch("watch_001") is True + assert db.get_watch("watch_001") is None + + def test_delete_nonexistent(self, db): + assert db.delete_watch("nope") is False + + +class TestWatchListQueries: + def test_list_for_ws(self, db): + db.create_watch(**_make_watch_kwargs(watch_id="w1", ws_id="ws-1", name="a")) + db.create_watch(**_make_watch_kwargs(watch_id="w2", ws_id="ws-1", name="b")) + db.create_watch(**_make_watch_kwargs(watch_id="w3", ws_id="ws-2", name="c")) + + ws1 = db.list_watches_for_ws("ws-1") + assert len(ws1) == 2 + assert {w["name"] for w in ws1} == {"a", "b"} + + def test_list_for_ws_excludes_inactive(self, db): + db.create_watch(**_make_watch_kwargs(watch_id="w1", ws_id="ws-1")) + db.update_watch("w1", active=False) + assert db.list_watches_for_ws("ws-1") == [] + + def test_list_for_node(self, db): + db.create_watch(**_make_watch_kwargs(watch_id="w1", node_id="n1")) + db.create_watch(**_make_watch_kwargs(watch_id="w2", node_id="n1")) + db.create_watch(**_make_watch_kwargs(watch_id="w3", node_id="n2")) + + n1 = db.list_watches_for_node("n1") + assert len(n1) == 2 + + def test_list_due(self, db): + # Due + db.create_watch(**_make_watch_kwargs(watch_id="w1", next_poll="2020-01-01T00:00:00")) + # Not due (far future) + db.create_watch(**_make_watch_kwargs(watch_id="w2", next_poll="2099-01-01T00:00:00")) + # Due but inactive + db.create_watch(**_make_watch_kwargs(watch_id="w3", next_poll="2020-01-01T00:00:00")) + db.update_watch("w3", active=False) + + due = db.list_due_watches("2025-01-01T00:00:00") + assert len(due) == 1 + assert due[0]["watch_id"] == "w1" + + def test_delete_for_ws(self, db): + db.create_watch(**_make_watch_kwargs(watch_id="w1", ws_id="ws-1")) + db.create_watch(**_make_watch_kwargs(watch_id="w2", ws_id="ws-1")) + db.create_watch(**_make_watch_kwargs(watch_id="w3", ws_id="ws-2")) + + count = db.delete_watches_for_ws("ws-1") + assert count == 2 + assert db.get_watch("w1") is None + assert db.get_watch("w2") is None + assert db.get_watch("w3") is not None diff --git a/turnstone/console/server.py b/turnstone/console/server.py index f5445309..9b91f7b1 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -1179,6 +1179,88 @@ async def admin_list_schedule_runs(request: Request) -> JSONResponse: return JSONResponse({"runs": runs}) +# --------------------------------------------------------------------------- +# Admin API endpoints — watches (aggregated from nodes) +# --------------------------------------------------------------------------- + + +async def admin_list_watches(request: Request) -> JSONResponse: + """GET /v1/api/admin/watches — aggregate watches from all nodes.""" + collector: ClusterCollector = request.app.state.collector + nodes, _ = collector.get_nodes(limit=500) + client: httpx.AsyncClient = request.app.state.proxy_client + headers = _proxy_auth_headers(request) + + async def _fetch_node(node: dict[str, Any]) -> list[dict[str, Any]]: + server_url = (node.get("server_url") or "").rstrip("/") + if not server_url: + return [] + try: + resp = await client.get(f"{server_url}/v1/api/watches", headers=headers) + if resp.status_code == 200: + data = resp.json() + watches: list[dict[str, Any]] = data.get("watches", []) + # Tag each watch with node_id in case the server omits it + for w in watches: + if not w.get("node_id"): + w["node_id"] = node["node_id"] + return watches + except Exception: + log.debug("Failed to fetch watches from node %s", node.get("node_id")) + return [] + + tasks = [_fetch_node(n) for n in nodes] + results = await asyncio.gather(*tasks) + all_watches: list[dict[str, Any]] = [] + for batch in results: + all_watches.extend(batch) + # Sort: active first, then by created descending (stable sort trick) + all_watches.sort(key=lambda w: w.get("created", ""), reverse=True) + all_watches.sort(key=lambda w: not w.get("active", False)) + return JSONResponse({"watches": all_watches}) + + +_VALID_WATCH_ID = re.compile(r"^[a-fA-F0-9]+$") + + +async def admin_cancel_watch(request: Request) -> Response: + """POST /v1/api/admin/watches/{watch_id}/cancel — proxy cancel to the owning node.""" + from turnstone.core.web_helpers import read_json_or_400 + + watch_id = request.path_params["watch_id"] + if not watch_id or not _VALID_WATCH_ID.match(watch_id) or len(watch_id) > 128: + return JSONResponse({"error": "Invalid watch_id"}, status_code=400) + + body = await read_json_or_400(request) + if isinstance(body, JSONResponse): + return body + + node_id = str(body.get("node_id", "") or request.query_params.get("node_id", "")).strip() + if not node_id: + return JSONResponse({"error": "node_id is required"}, status_code=400) + + server_url = _get_server_url(request, node_id) + if not server_url: + return JSONResponse({"error": "Node not found"}, status_code=404) + + client: httpx.AsyncClient = request.app.state.proxy_client + headers = {"Content-Type": "application/json"} + headers.update(_proxy_auth_headers(request)) + try: + resp = await client.post( + f"{server_url}/v1/api/watches/{watch_id}/cancel", + content=b"{}", + headers=headers, + ) + return Response( + content=resp.content, + status_code=resp.status_code, + media_type=resp.headers.get("content-type", "application/json"), + ) + except httpx.HTTPError: + return JSONResponse({"error": "Node unreachable"}, status_code=502) + + # --------------------------------------------------------------------------- # App factory # --------------------------------------------------------------------------- @@ -1249,6 +1331,12 @@ def create_app( methods=["DELETE"], ), Route("/api/admin/schedules/{task_id}/runs", admin_list_schedule_runs), + Route("/api/admin/watches", admin_list_watches), + Route( + "/api/admin/watches/{watch_id}/cancel", + admin_cancel_watch, + methods=["POST"], + ), ], ), Route("/health", health), diff --git a/turnstone/console/static/admin.js b/turnstone/console/static/admin.js index c7f45c20..a98b4e97 100644 --- a/turnstone/console/static/admin.js +++ b/turnstone/console/static/admin.js @@ -10,6 +10,7 @@ var _ctTrapHandler = null; var _tcTrapHandler = null; var _ccTrapHandler = null; var _cfTrapHandler = null; +var _adminWatches = []; var _confirmCallbackFn = null; var _confirmTriggerEl = null; @@ -48,11 +49,14 @@ function switchAdminTab(tab) { tab === "channels" ? "" : "none"; document.getElementById("admin-schedules").style.display = tab === "schedules" ? "" : "none"; + document.getElementById("admin-watches").style.display = + tab === "watches" ? "" : "none"; if (tab === "users") loadAdminUsers(); if (tab === "tokens") _populateTokenUserSelect(); if (tab === "channels") _populateChannelUserSelect(); if (tab === "schedules") loadAdminSchedules(); + if (tab === "watches") loadAdminWatches(); } // --------------------------------------------------------------------------- @@ -887,6 +891,167 @@ function hideScheduleRunsModal() { _runsScheduleTriggerEl = null; } +// --------------------------------------------------------------------------- +// Watches +// --------------------------------------------------------------------------- + +function _populateWatchNodeSelect() { + var sel = document.getElementById("admin-watch-node"); + var current = sel.value; + var seen = {}; + sel.innerHTML = ''; + for (var i = 0; i < _adminWatches.length; i++) { + var nid = _adminWatches[i].node_id || ""; + if (nid && !seen[nid]) { + seen[nid] = true; + var opt = document.createElement("option"); + opt.value = nid; + opt.textContent = nid; + sel.appendChild(opt); + } + } + if (current) sel.value = current; +} + +function loadAdminWatches() { + authFetch("/v1/api/admin/watches") + .then(function (r) { + if (!r.ok) throw new Error("Failed to load watches"); + return r.json(); + }) + .then(function (data) { + _adminWatches = data.watches || []; + _populateWatchNodeSelect(); + var nodeFilter = document.getElementById("admin-watch-node").value; + var filtered = _adminWatches; + if (nodeFilter) { + filtered = _adminWatches.filter(function (w) { + return w.node_id === nodeFilter; + }); + } + _renderWatches(filtered); + }) + .catch(function () { + document.getElementById("admin-watches-table").innerHTML = + '
Failed to load watches
'; + }); +} + +function _formatInterval(secs) { + if (!secs || secs <= 0) return "\u2014"; + if (secs >= 3600) return Math.round(secs / 3600) + "h"; + if (secs >= 60) return Math.round(secs / 60) + "m"; + return secs + "s"; +} + +function _renderWatches(watches) { + var container = document.getElementById("admin-watches-table"); + if (!watches.length) { + container.innerHTML = + '
No active watches. Watches are created when workstreams use the watch tool.
'; + return; + } + var html = ""; + for (var i = 0; i < watches.length; i++) { + var w = watches[i]; + var name = w.name || w.watch_id || "\u2014"; + var nodeShort = (w.node_id || "").slice(0, 8); + var cmd = w.command || ""; + var cmdTrunc = cmd.length > 40 ? cmd.slice(0, 40) + "\u2026" : cmd; + var interval = _formatInterval(w.interval_secs); + var pollMax = w.max_polls ? w.max_polls : "\u221e"; + var pollLabel = (w.poll_count || 0) + "/" + pollMax; + var cond = w.stop_on || "on change"; + var condTrunc = cond.length > 30 ? cond.slice(0, 30) + "\u2026" : cond; + var active = w.active; + var statusCls = active ? "watch-active" : "watch-completed"; + var statusLabel = active ? "active" : "done"; + var statusDot = active ? "\u25cf " : "\u25cb "; + var cancelBtn = active + ? '' + : ""; + html += + '
' + + '' + + escapeHtml(name) + + "" + + '' + + escapeHtml(nodeShort) + + "" + + '' + + escapeHtml(cmdTrunc) + + "" + + '' + + escapeHtml(interval) + + "" + + '' + + escapeHtml(pollLabel) + + "" + + '' + + escapeHtml(condTrunc) + + "" + + '' + + statusDot + + statusLabel + + "" + + '' + + cancelBtn + + "
"; + } + container.innerHTML = html; + // Bind cancel buttons + var btns = container.querySelectorAll("[data-cancel-watch]"); + for (var j = 0; j < btns.length; j++) { + btns[j].addEventListener("click", function () { + _cancelWatch( + this.getAttribute("data-cancel-watch"), + this.getAttribute("data-watch-node"), + this.getAttribute("data-watch-name"), + ); + }); + } +} + +function _cancelWatch(watchId, nodeId, name) { + showConfirmModal( + "Cancel Watch", + "Cancel watch \u2018" + name + "\u2019? This will stop future polling.", + "Cancel watch", + function () { + authFetch( + "/v1/api/admin/watches/" + encodeURIComponent(watchId) + "/cancel", + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ node_id: nodeId }), + }, + ) + .then(function (r) { + if (!r.ok) throw new Error("Cancel failed"); + showToast("Watch '" + name + "' cancelled"); + loadAdminWatches(); + }) + .catch(function () { + showToast("Failed to cancel watch"); + }); + }, + ); +} + // --------------------------------------------------------------------------- // Create Channel Link Modal // --------------------------------------------------------------------------- @@ -1271,7 +1436,7 @@ document.addEventListener("keydown", function (e) { if (!tablist) return; tablist.addEventListener("keydown", function (e) { if (e.key !== "ArrowLeft" && e.key !== "ArrowRight") return; - var tabOrder = ["users", "tokens", "channels", "schedules"]; + var tabOrder = ["users", "tokens", "channels", "schedules", "watches"]; var idx = tabOrder.indexOf(_adminTab); if (e.key === "ArrowRight") idx = (idx + 1) % tabOrder.length; else idx = (idx - 1 + tabOrder.length) % tabOrder.length; diff --git a/turnstone/console/static/index.html b/turnstone/console/static/index.html index 550babcf..d5beff84 100644 --- a/turnstone/console/static/index.html +++ b/turnstone/console/static/index.html @@ -81,6 +81,7 @@ + @@ -163,6 +164,30 @@
Loading schedules...
+ + + diff --git a/turnstone/console/static/style.css b/turnstone/console/static/style.css index b34ac4d5..2eec3442 100644 --- a/turnstone/console/static/style.css +++ b/turnstone/console/static/style.css @@ -863,6 +863,16 @@ .sched-disabled { color: var(--fg-dim); } .sched-expired { color: var(--accent); } +/* Watches grid: NAME | NODE | COMMAND | INTERVAL | POLL | CONDITION | STATUS | ACTIONS */ +#admin-watches .admin-colheaders, +#admin-watches .admin-row { + grid-template-columns: 1.2fr 80px 1.5fr 60px 70px 1fr 70px 70px; +} + +/* Watch status indicators */ +.watch-active { color: var(--green); font-weight: 500; } +.watch-completed { color: var(--accent); } + /* Wide modal variant for schedule forms */ .admin-modal-wide { width: 480px; } @@ -1027,6 +1037,10 @@ grid-template-columns: 1fr 60px 80px 130px; } .admin-col-sschedule, .admin-col-starget, .admin-col-snext { display: none; } + #admin-watches .admin-colheaders, #admin-watches .admin-row { + grid-template-columns: 1.2fr 80px 70px 70px 70px; + } + .admin-col-wcmd, .admin-col-wcond, .admin-col-winterval { display: none; } } /* ========================================================================== diff --git a/turnstone/core/auth.py b/turnstone/core/auth.py index 57c9a114..d456f577 100644 --- a/turnstone/core/auth.py +++ b/turnstone/core/auth.py @@ -390,6 +390,13 @@ def required_scope(method: str, path: str) -> str: # Write endpoints if method == "POST" and normalized in WRITE_PATHS: return "write" + # Watch cancel has a path parameter: /api/watches/{id}/cancel + if ( + method == "POST" + and normalized.startswith("/api/watches/") + and normalized.endswith("/cancel") + ): + return "write" # Console proxy routes: /node/{node_id}/api/{tail} or /node/{node_id}/v1/api/{tail} if method == "POST" and normalized.startswith("/node/"): diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 2d4d045b..92cf82e0 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -15,6 +15,7 @@ import dataclasses import json import mimetypes import os +import queue import re import signal import subprocess @@ -222,6 +223,10 @@ class ChatSession: self._assistant_pending_tokens = 0 self.creative_mode = False self._notify_count = 0 + # Watch support: server-level runner injected via set_watch_runner() + self._watch_runner: Any = None # WatchRunner | None + self._watch_pending: queue.Queue[dict[str, Any]] = queue.Queue() + self._watch_dispatch_depth = 0 # MCP tool integration: merge external tools with built-in self._mcp_client = mcp_client self._mcp_refresh_cb: Any = None # Callable | None (avoid import) @@ -330,11 +335,32 @@ class ChatSession: else: self._tool_search = None + def set_watch_runner(self, runner: Any, dispatch_fn: Any = None) -> None: + """Inject the server-level WatchRunner (called after workstream setup). + + If *dispatch_fn* is provided (the server passes one that can start + worker threads), it is registered directly. Otherwise a simple + enqueue fallback is used — suitable only when ``send()`` is already + active (Path A). + """ + self._watch_runner = runner + if dispatch_fn is not None: + runner.set_dispatch_fn(self._ws_id, dispatch_fn) + else: + pending = self._watch_pending + + def _enqueue(msg: str) -> None: + pending.put({"message": msg}) + + runner.set_dispatch_fn(self._ws_id, _enqueue) + def close(self) -> None: """Release resources (listener registrations, etc.).""" if self._mcp_client and self._mcp_refresh_cb: self._mcp_client.remove_listener(self._mcp_refresh_cb) self._mcp_refresh_cb = None + if self._watch_runner: + self._watch_runner.remove_dispatch_fn(self._ws_id) def _handle_mcp_refresh(self, arg: str) -> None: """Handle ``/mcp refresh [server]``.""" @@ -766,6 +792,9 @@ class ChatSession: self._title_generated = True threading.Thread(target=self._generate_title, daemon=True).start() self._emit_state("idle") + # Dispatch any pending watch results (chains into + # a new send() within the same worker thread). + self._dispatch_pending_watch(self._watch_dispatch_depth) break # Execute tool calls (potentially in parallel) @@ -1520,6 +1549,7 @@ class ChatSession: "recall": self._prepare_recall, "forget": self._prepare_forget, "notify": self._prepare_notify, + "watch": self._prepare_watch, } preparer = preparers.get(func_name) if not preparer: @@ -2957,6 +2987,289 @@ class ChatSession: self.ui.on_tool_result(call_id, "notify", msg) return call_id, msg + # -- Watch tool ---------------------------------------------------------- + + def _prepare_watch(self, call_id: str, args: dict[str, Any]) -> dict[str, Any]: + from turnstone.core.watch import ( + MAX_INTERVAL, + MAX_WATCHES_PER_WS, + MIN_INTERVAL, + parse_duration, + validate_condition, + ) + + action = args.get("action", "") + if action == "list": + return { + "call_id": call_id, + "func_name": "watch", + "header": "\u23f1 watch: list", + "preview": "", + "needs_approval": False, + "execute": self._exec_watch, + "action": "list", + } + if action == "cancel": + name = args.get("name", "") + if not name: + return { + "call_id": call_id, + "func_name": "watch", + "header": "\u2717 watch cancel: missing name", + "preview": "", + "needs_approval": False, + "error": "Error: 'name' is required for cancel", + } + return { + "call_id": call_id, + "func_name": "watch", + "header": f'\u23f1 watch: cancel "{name}"', + "preview": "", + "needs_approval": False, + "execute": self._exec_watch, + "action": "cancel", + "watch_name": name, + } + if action != "create": + return { + "call_id": call_id, + "func_name": "watch", + "header": f"\u2717 watch: unknown action '{action}'", + "preview": "", + "needs_approval": False, + "error": f"Error: unknown action '{action}'. Use create, list, or cancel.", + } + + # --- action=create --- + command = sanitize_command(args.get("command", "")) + if not command: + return { + "call_id": call_id, + "func_name": "watch", + "header": "\u2717 watch create: missing command", + "preview": "", + "needs_approval": False, + "error": "Error: 'command' is required for create", + } + blocked = is_command_blocked(command) + if blocked: + return { + "call_id": call_id, + "func_name": "watch", + "header": f"\u2717 {blocked}", + "preview": "", + "needs_approval": False, + "error": blocked, + } + + # Parse poll interval + poll_every_str = args.get("poll_every", "5m") + try: + interval_secs = parse_duration(poll_every_str) + except ValueError as exc: + return { + "call_id": call_id, + "func_name": "watch", + "header": f"\u2717 watch: invalid poll_every: {exc}", + "preview": "", + "needs_approval": False, + "error": f"Error: invalid poll_every: {exc}", + } + if interval_secs < MIN_INTERVAL: + return { + "call_id": call_id, + "func_name": "watch", + "header": f"\u2717 watch: interval too short (min {MIN_INTERVAL}s)", + "preview": "", + "needs_approval": False, + "error": f"Error: minimum poll interval is {MIN_INTERVAL}s", + } + if interval_secs > MAX_INTERVAL: + return { + "call_id": call_id, + "func_name": "watch", + "header": f"\u2717 watch: interval too long (max {MAX_INTERVAL}s)", + "preview": "", + "needs_approval": False, + "error": f"Error: maximum poll interval is {MAX_INTERVAL}s", + } + + # Validate stop condition + stop_on = args.get("stop_on") + if stop_on is not None: + err = validate_condition(stop_on) + if err: + return { + "call_id": call_id, + "func_name": "watch", + "header": f"\u2717 watch: {err}", + "preview": "", + "needs_approval": False, + "error": f"Error: {err}", + } + + # Check max watches limit and duplicate names + storage = get_storage() + existing: list[dict[str, Any]] = [] + if storage: + existing = storage.list_watches_for_ws(self._ws_id) + if len(existing) >= MAX_WATCHES_PER_WS: + return { + "call_id": call_id, + "func_name": "watch", + "header": f"\u2717 watch: limit reached ({MAX_WATCHES_PER_WS})", + "preview": "", + "needs_approval": False, + "error": f"Error: maximum {MAX_WATCHES_PER_WS} active watches per workstream", + } + + name = args.get("name", "") + if not name: + name = f"watch-{uuid.uuid4().hex[:4]}" + elif storage and any(w["name"] == name for w in existing): + return { + "call_id": call_id, + "func_name": "watch", + "header": f'\u2717 watch: name "{name}" already in use', + "preview": "", + "needs_approval": False, + "error": f'Error: a watch named "{name}" already exists in this workstream', + } + max_polls = args.get("max_polls", 100) + try: + max_polls = int(max_polls) + except (ValueError, TypeError): + max_polls = 100 + + display_cmd = command.split("\n")[0] + condition_display = f", stop_on={stop_on}" if stop_on else ", on change" + return { + "call_id": call_id, + "func_name": "watch", + "header": f'\u23f1 watch: "{name}" every {poll_every_str}', + "preview": f" {display_cmd}{condition_display}", + "needs_approval": True, + "approval_label": "watch", + "execute": self._exec_watch, + "action": "create", + "command": command, + "interval_secs": interval_secs, + "stop_on": stop_on, + "watch_name": name, + "max_polls": max_polls, + } + + def _exec_watch(self, item: dict[str, Any]) -> tuple[str, str]: + from datetime import UTC, datetime, timedelta + + call_id = item["call_id"] + action = item["action"] + storage = get_storage() + + if action == "list": + if not storage: + msg = "No watches (storage unavailable)" + self.ui.on_tool_result(call_id, "watch", msg) + return call_id, msg + watches = storage.list_watches_for_ws(self._ws_id) + if not watches: + msg = "No active watches." + self.ui.on_tool_result(call_id, "watch", msg) + return call_id, msg + from turnstone.core.watch import format_interval + + lines = [] + for w in watches: + condition = w.get("stop_on") or "on change" + lines.append( + f" {w['name']} ({w['watch_id'][:8]}): " + f"every {format_interval(w['interval_secs'])}, " + f"poll #{w['poll_count']}/{w['max_polls']}, " + f"condition: {condition}, " + f"cmd: {w['command'][:60]}" + ) + msg = "Active watches:\n" + "\n".join(lines) + self.ui.on_tool_result(call_id, "watch", msg) + return call_id, msg + + if action == "cancel": + name = item.get("watch_name", "") + if not storage: + msg = "Error: storage unavailable" + self.ui.on_tool_result(call_id, "watch", msg) + return call_id, msg + watches = storage.list_watches_for_ws(self._ws_id) + target = None + for w in watches: + if w["name"] == name or w["watch_id"].startswith(name): + target = w + break + if target is None: + msg = f'Watch "{name}" not found.' + self.ui.on_tool_result(call_id, "watch", msg) + return call_id, msg + storage.update_watch(target["watch_id"], active=False, next_poll="") + msg = f'Watch "{target["name"]}" cancelled.' + self.ui.on_tool_result(call_id, "watch", msg) + return call_id, msg + + # action == "create" + if not storage: + msg = "Error: storage unavailable" + self.ui.on_tool_result(call_id, "watch", msg) + return call_id, msg + + watch_id = uuid.uuid4().hex + now = datetime.now(UTC) + next_poll = now + timedelta(seconds=item["interval_secs"]) + storage.create_watch( + watch_id=watch_id, + ws_id=self._ws_id, + node_id=self._node_id or "", + name=item["watch_name"], + command=item["command"], + interval_secs=item["interval_secs"], + stop_on=item.get("stop_on"), + max_polls=item["max_polls"], + created_by="model", + next_poll=next_poll.strftime("%Y-%m-%dT%H:%M:%S"), + ) + + from turnstone.core.watch import format_interval + + stop_desc = f"stop_on: {item['stop_on']}" if item.get("stop_on") else "on output change" + msg = ( + f'Watch "{item["watch_name"]}" created.\n' + f" Polling every {format_interval(item['interval_secs'])}, " + f"max {item['max_polls']} polls\n" + f" Command: {item['command']}\n" + f" Condition: {stop_desc}" + ) + self.ui.on_tool_result(call_id, "watch", msg) + return call_id, msg + + _MAX_WATCH_CHAIN = 5 # max consecutive watch dispatches per worker thread + + def _dispatch_pending_watch(self, depth: int = 0) -> None: + """Dispatch one pending watch result as a new send() turn. + + Each ``send()`` chains back here on IDLE, so multiple queued results + are processed sequentially. Depth is capped to prevent unbounded + stack growth. + """ + if depth >= self._MAX_WATCH_CHAIN: + self._watch_dispatch_depth = 0 + return + try: + result = self._watch_pending.get_nowait() + except queue.Empty: + self._watch_dispatch_depth = 0 # chain ended — reset for next user turn + return + message = result.get("message", "") + if message: + self._watch_dispatch_depth = depth + 1 + self.send(message) + def _exec_write_file(self, item: dict[str, Any]) -> tuple[str, str]: """Write content to a file, creating parent directories as needed.""" call_id = item["call_id"] diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py index 01a54bf5..923b5dbd 100644 --- a/turnstone/core/storage/_postgresql.py +++ b/turnstone/core/storage/_postgresql.py @@ -1011,6 +1011,139 @@ class PostgreSQLBackend: conn.commit() return result.rowcount + # -- Watches --------------------------------------------------------------- + + def create_watch( + self, + watch_id: str, + ws_id: str, + node_id: str, + name: str, + command: str, + interval_secs: float, + stop_on: str | None, + max_polls: int, + created_by: str, + next_poll: str, + ) -> None: + from sqlalchemy.dialects import postgresql + + from turnstone.core.storage._schema import watches + + now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + with self._engine.connect() as conn: + conn.execute( + postgresql.insert(watches) + .values( + watch_id=watch_id, + ws_id=ws_id, + node_id=node_id, + name=name, + command=command, + interval_secs=interval_secs, + stop_on=stop_on, + max_polls=max_polls, + poll_count=0, + active=1, + created_by=created_by, + next_poll=next_poll, + created=now, + updated=now, + ) + .on_conflict_do_nothing() + ) + conn.commit() + + def get_watch(self, watch_id: str) -> dict[str, Any] | None: + from turnstone.core.storage._schema import watches + + with self._engine.connect() as conn: + row = conn.execute(sa.select(watches).where(watches.c.watch_id == watch_id)).fetchone() + if row is None: + return None + return dict(row._mapping) + + def list_watches_for_ws(self, ws_id: str) -> list[dict[str, Any]]: + from turnstone.core.storage._schema import watches + + with self._engine.connect() as conn: + rows = conn.execute( + sa.select(watches) + .where((watches.c.ws_id == ws_id) & (watches.c.active == 1)) + .order_by(watches.c.created.desc()) + ).fetchall() + return [dict(r._mapping) for r in rows] + + def list_watches_for_node(self, node_id: str) -> list[dict[str, Any]]: + from turnstone.core.storage._schema import watches + + with self._engine.connect() as conn: + rows = conn.execute( + sa.select(watches) + .where((watches.c.node_id == node_id) & (watches.c.active == 1)) + .order_by(watches.c.created.desc()) + ).fetchall() + return [dict(r._mapping) for r in rows] + + def list_due_watches(self, now: str) -> list[dict[str, Any]]: + from turnstone.core.storage._schema import watches + + with self._engine.connect() as conn: + rows = conn.execute( + sa.select(watches) + .where( + (watches.c.active == 1) + & (watches.c.next_poll <= now) + & (watches.c.next_poll != "") + ) + .order_by(watches.c.next_poll) + .limit(100) + ).fetchall() + return [dict(r._mapping) for r in rows] + + _UPDATABLE_WATCH_FIELDS = frozenset( + { + "name", + "poll_count", + "last_output", + "last_exit_code", + "last_poll", + "next_poll", + "active", + "updated", + } + ) + + def update_watch(self, watch_id: str, **fields: Any) -> bool: + from turnstone.core.storage._schema import watches + + fields = {k: v for k, v in fields.items() if k in self._UPDATABLE_WATCH_FIELDS} + fields["updated"] = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + if "active" in fields: + fields["active"] = 1 if fields["active"] else 0 + with self._engine.connect() as conn: + result = conn.execute( + sa.update(watches).where(watches.c.watch_id == watch_id).values(**fields) + ) + conn.commit() + return result.rowcount > 0 + + def delete_watch(self, watch_id: str) -> bool: + from turnstone.core.storage._schema import watches + + with self._engine.connect() as conn: + result = conn.execute(sa.delete(watches).where(watches.c.watch_id == watch_id)) + conn.commit() + return result.rowcount > 0 + + def delete_watches_for_ws(self, ws_id: str) -> int: + from turnstone.core.storage._schema import watches + + with self._engine.connect() as conn: + result = conn.execute(sa.delete(watches).where(watches.c.ws_id == ws_id)) + conn.commit() + return result.rowcount + # -- Service registry ------------------------------------------------------ def register_service( diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py index 2dbf6d46..8fa9fd26 100644 --- a/turnstone/core/storage/_protocol.py +++ b/turnstone/core/storage/_protocol.py @@ -293,6 +293,52 @@ class StorageBackend(Protocol): """Delete task runs older than retention_days. Returns count deleted.""" ... + # -- Watches --------------------------------------------------------------- + + def create_watch( + self, + watch_id: str, + ws_id: str, + node_id: str, + name: str, + command: str, + interval_secs: float, + stop_on: str | None, + max_polls: int, + created_by: str, + next_poll: str, + ) -> None: + """Create a watch. No-op if watch_id already exists.""" + ... + + def get_watch(self, watch_id: str) -> dict[str, Any] | None: + """Return watch dict or None.""" + ... + + def list_watches_for_ws(self, ws_id: str) -> list[dict[str, Any]]: + """Return active watches for a workstream, ordered by created DESC.""" + ... + + def list_watches_for_node(self, node_id: str) -> list[dict[str, Any]]: + """Return all active watches on a node, ordered by created DESC.""" + ... + + def list_due_watches(self, now: str) -> list[dict[str, Any]]: + """Return active watches whose next_poll <= now, ordered by next_poll.""" + ... + + def update_watch(self, watch_id: str, **fields: Any) -> bool: + """Update specified fields on a watch. Returns True if found.""" + ... + + def delete_watch(self, watch_id: str) -> bool: + """Delete a watch. Returns True if found.""" + ... + + def delete_watches_for_ws(self, ws_id: str) -> int: + """Delete all watches for a workstream. Returns count deleted.""" + ... + # -- Service registry ------------------------------------------------------ def register_service( diff --git a/turnstone/core/storage/_schema.py b/turnstone/core/storage/_schema.py index e3702776..84392225 100644 --- a/turnstone/core/storage/_schema.py +++ b/turnstone/core/storage/_schema.py @@ -170,6 +170,40 @@ sa.Index("idx_scheduled_task_runs_started", scheduled_task_runs.c.started) # Service registry # --------------------------------------------------------------------------- +# --------------------------------------------------------------------------- +# Watches — in-session periodic command polling +# --------------------------------------------------------------------------- + +watches = sa.Table( + "watches", + metadata, + sa.Column("watch_id", sa.Text, primary_key=True), + sa.Column("ws_id", sa.Text, nullable=False), + sa.Column("node_id", sa.Text, nullable=False, server_default=""), + sa.Column("name", sa.Text, nullable=False), + sa.Column("command", sa.Text, nullable=False), + sa.Column("interval_secs", sa.Float, nullable=False), + sa.Column("stop_on", sa.Text), # Python expression, NULL = change detection + sa.Column("max_polls", sa.Integer, nullable=False, server_default="100"), + sa.Column("poll_count", sa.Integer, nullable=False, server_default="0"), + sa.Column("last_output", sa.Text), + sa.Column("last_exit_code", sa.Integer), + sa.Column("last_poll", sa.Text), # ISO8601 + sa.Column("next_poll", sa.Text), # ISO8601 + sa.Column("active", sa.Integer, nullable=False, server_default="1"), + sa.Column("created_by", sa.Text, nullable=False, server_default=""), + sa.Column("created", sa.Text, nullable=False), + sa.Column("updated", sa.Text, nullable=False), +) + +sa.Index("idx_watches_active_next", watches.c.active, watches.c.next_poll) +sa.Index("idx_watches_ws_id", watches.c.ws_id) +sa.Index("idx_watches_node_id", watches.c.node_id) + +# --------------------------------------------------------------------------- +# Service registry +# --------------------------------------------------------------------------- + services = sa.Table( "services", metadata, diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py index 10a7fdf5..6b8f2461 100644 --- a/turnstone/core/storage/_sqlite.py +++ b/turnstone/core/storage/_sqlite.py @@ -1062,6 +1062,136 @@ class SQLiteBackend: conn.commit() return result.rowcount + # -- Watches --------------------------------------------------------------- + + def create_watch( + self, + watch_id: str, + ws_id: str, + node_id: str, + name: str, + command: str, + interval_secs: float, + stop_on: str | None, + max_polls: int, + created_by: str, + next_poll: str, + ) -> None: + from turnstone.core.storage._schema import watches + + now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + with self._engine.connect() as conn: + conn.execute( + sa.insert(watches).prefix_with("OR IGNORE"), + { + "watch_id": watch_id, + "ws_id": ws_id, + "node_id": node_id, + "name": name, + "command": command, + "interval_secs": interval_secs, + "stop_on": stop_on, + "max_polls": max_polls, + "poll_count": 0, + "active": 1, + "created_by": created_by, + "next_poll": next_poll, + "created": now, + "updated": now, + }, + ) + conn.commit() + + def get_watch(self, watch_id: str) -> dict[str, Any] | None: + from turnstone.core.storage._schema import watches + + with self._engine.connect() as conn: + row = conn.execute(sa.select(watches).where(watches.c.watch_id == watch_id)).fetchone() + if row is None: + return None + return dict(row._mapping) + + def list_watches_for_ws(self, ws_id: str) -> list[dict[str, Any]]: + from turnstone.core.storage._schema import watches + + with self._engine.connect() as conn: + rows = conn.execute( + sa.select(watches) + .where((watches.c.ws_id == ws_id) & (watches.c.active == 1)) + .order_by(watches.c.created.desc()) + ).fetchall() + return [dict(r._mapping) for r in rows] + + def list_watches_for_node(self, node_id: str) -> list[dict[str, Any]]: + from turnstone.core.storage._schema import watches + + with self._engine.connect() as conn: + rows = conn.execute( + sa.select(watches) + .where((watches.c.node_id == node_id) & (watches.c.active == 1)) + .order_by(watches.c.created.desc()) + ).fetchall() + return [dict(r._mapping) for r in rows] + + def list_due_watches(self, now: str) -> list[dict[str, Any]]: + from turnstone.core.storage._schema import watches + + with self._engine.connect() as conn: + rows = conn.execute( + sa.select(watches) + .where( + (watches.c.active == 1) + & (watches.c.next_poll <= now) + & (watches.c.next_poll != "") + ) + .order_by(watches.c.next_poll) + .limit(100) + ).fetchall() + return [dict(r._mapping) for r in rows] + + _UPDATABLE_WATCH_FIELDS = frozenset( + { + "name", + "poll_count", + "last_output", + "last_exit_code", + "last_poll", + "next_poll", + "active", + "updated", + } + ) + + def update_watch(self, watch_id: str, **fields: Any) -> bool: + from turnstone.core.storage._schema import watches + + fields = {k: v for k, v in fields.items() if k in self._UPDATABLE_WATCH_FIELDS} + fields["updated"] = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + if "active" in fields: + fields["active"] = 1 if fields["active"] else 0 + with self._engine.connect() as conn: + result = conn.execute( + sa.update(watches).where(watches.c.watch_id == watch_id).values(**fields) + ) + conn.commit() + return result.rowcount > 0 + + def delete_watch(self, watch_id: str) -> bool: + from turnstone.core.storage._schema import watches + + with self._engine.connect() as conn: + result = conn.execute(sa.delete(watches).where(watches.c.watch_id == watch_id)) + conn.commit() + return result.rowcount > 0 + + def delete_watches_for_ws(self, ws_id: str) -> int: + from turnstone.core.storage._schema import watches + + with self._engine.connect() as conn: + result = conn.execute(sa.delete(watches).where(watches.c.ws_id == ws_id)) + conn.commit() + return result.rowcount + # -- Service registry ------------------------------------------------------ def register_service( diff --git a/turnstone/core/storage/migrations/versions/007_watches.py b/turnstone/core/storage/migrations/versions/007_watches.py new file mode 100644 index 00000000..c35398e5 --- /dev/null +++ b/turnstone/core/storage/migrations/versions/007_watches.py @@ -0,0 +1,47 @@ +"""Watches table for in-session periodic command polling. + +Revision ID: 007 +Revises: 006 +Create Date: 2026-03-09 +""" + +import sqlalchemy as sa +from alembic import op + +revision = "007" +down_revision = "006" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "watches", + sa.Column("watch_id", sa.Text, primary_key=True), + sa.Column("ws_id", sa.Text, nullable=False), + sa.Column("node_id", sa.Text, nullable=False, server_default=""), + sa.Column("name", sa.Text, nullable=False), + sa.Column("command", sa.Text, nullable=False), + sa.Column("interval_secs", sa.Float, nullable=False), + sa.Column("stop_on", sa.Text), + sa.Column("max_polls", sa.Integer, nullable=False, server_default="100"), + sa.Column("poll_count", sa.Integer, nullable=False, server_default="0"), + sa.Column("last_output", sa.Text), + sa.Column("last_exit_code", sa.Integer), + sa.Column("last_poll", sa.Text), + sa.Column("next_poll", sa.Text), + sa.Column("active", sa.Integer, nullable=False, server_default="1"), + sa.Column("created_by", sa.Text, nullable=False, server_default=""), + sa.Column("created", sa.Text, nullable=False), + sa.Column("updated", sa.Text, nullable=False), + ) + op.create_index("idx_watches_active_next", "watches", ["active", "next_poll"]) + op.create_index("idx_watches_ws_id", "watches", ["ws_id"]) + op.create_index("idx_watches_node_id", "watches", ["node_id"]) + + +def downgrade() -> None: + op.drop_index("idx_watches_node_id", "watches") + op.drop_index("idx_watches_ws_id", "watches") + op.drop_index("idx_watches_active_next", "watches") + op.drop_table("watches") diff --git a/turnstone/core/watch.py b/turnstone/core/watch.py new file mode 100644 index 00000000..a1cd9f12 --- /dev/null +++ b/turnstone/core/watch.py @@ -0,0 +1,442 @@ +"""Watch — periodic command polling within a workstream. + +A watch periodically runs a shell command and injects results back into the +conversation when a stop condition is met or the output changes. The +``WatchRunner`` is a server-level daemon thread that polls the database for +due watches, runs their commands, and dispatches results. +""" + +from __future__ import annotations + +import contextlib +import json +import logging +import re +import subprocess +import threading +from datetime import UTC, datetime, timedelta +from typing import TYPE_CHECKING, Any + +from turnstone.core.safety import is_command_blocked, sanitize_command + +if TYPE_CHECKING: + from collections.abc import Callable + +log = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +MAX_WATCHES_PER_WS = 5 +MIN_INTERVAL = 10 # seconds +MAX_INTERVAL = 86_400 # 24 hours +DEFAULT_MAX_POLLS = 100 +DEFAULT_INTERVAL = 300 # 5 minutes +MAX_OUTPUT_SIZE = 65_536 # truncate stored/dispatched output at 64 KB + +# Safe builtins exposed to condition expressions. +_SAFE_BUILTINS: dict[str, Any] = { + "len": len, + "str": str, + "int": int, + "float": float, + "bool": bool, + "abs": abs, + "min": min, + "max": max, + "any": any, + "all": all, + "isinstance": isinstance, + "sorted": sorted, + "True": True, + "False": False, + "None": None, +} + +# --------------------------------------------------------------------------- +# Duration parsing +# --------------------------------------------------------------------------- + +_DURATION_RE = re.compile(r"(?:(\d+)\s*h)?\s*(?:(\d+)\s*m)?\s*(?:(\d+)\s*s)?$", re.IGNORECASE) + + +def parse_duration(s: str) -> float: + """Convert a duration string to seconds. + + Supported formats: ``"30s"``, ``"5m"``, ``"1h"``, ``"2h30m"``, + ``"90"`` (bare number = seconds). + + Raises ``ValueError`` on invalid input. + """ + s = s.strip() + if not s: + raise ValueError("empty duration string") + + # Bare number → seconds + try: + val = float(s) + except ValueError: + val = None + if val is not None: + if val <= 0: + raise ValueError(f"duration must be positive, got {val}") + return val + + m = _DURATION_RE.match(s) + if not m or not any(m.groups()): + raise ValueError(f"invalid duration format: {s!r}") + + hours = int(m.group(1) or 0) + minutes = int(m.group(2) or 0) + seconds = int(m.group(3) or 0) + total = hours * 3600 + minutes * 60 + seconds + if total <= 0: + raise ValueError(f"duration must be positive, got {total}s") + return float(total) + + +# --------------------------------------------------------------------------- +# Condition evaluation +# --------------------------------------------------------------------------- + + +def validate_condition(expr: str) -> str | None: + """Syntax-check a condition expression. + + Returns an error message string, or ``None`` if the expression is valid. + """ + try: + compile(expr, "", "eval") + except SyntaxError as exc: + return f"invalid condition syntax: {exc}" + return None + + +def evaluate_condition( + expr: str | None, + output: str, + exit_code: int, + prev_output: str | None, +) -> tuple[bool, str]: + """Evaluate a stop condition. + + Returns ``(fired, reason)`` where *fired* is ``True`` when the watch + should report a result and *reason* is a human-readable explanation. + """ + changed = output != prev_output + + if expr is None: + # Default: fire on any change (skip first poll where prev is None) + if prev_output is None: + return False, "" + return changed, "output changed" if changed else "" + + # Build data context + data: Any = None + with contextlib.suppress(json.JSONDecodeError, ValueError): + data = json.loads(output) + + context = { + "output": output, + "data": data, + "exit_code": exit_code, + "prev_output": prev_output, + "changed": changed, + } + + try: + result = eval(expr, {"__builtins__": _SAFE_BUILTINS}, context) # noqa: S307 + if result: + return True, f"condition met: {expr}" + return False, "" + except Exception as exc: + log.warning("watch.condition_error", extra={"expr": expr, "error": str(exc)}) + return False, f"condition error: {exc}" + + +# --------------------------------------------------------------------------- +# Message formatting +# --------------------------------------------------------------------------- + + +def format_watch_message( + name: str, + command: str, + output: str, + poll_count: int, + max_polls: int, + elapsed_secs: float, + stop_on: str | None, + is_final: bool, + reason: str, +) -> str: + """Format a watch result as a synthetic user message.""" + elapsed = format_interval(elapsed_secs) + lines = [f'[Watch "{name}" \u2014 poll #{poll_count}/{max_polls}, {elapsed} elapsed]'] + + # Show the condition so the model knows what this watch was waiting for + if stop_on: + lines.append(f"[condition: {stop_on}]") + else: + lines.append("[mode: fire on output change]") + + lines.append("") + lines.append(f"$ {command}") + lines.append(output) + + if is_final: + if reason: + lines.append("") + lines.append(f"[{reason} \u2014 watch auto-cancelled]") + else: + lines.append("") + lines.append("[max polls reached \u2014 watch auto-cancelled]") + + return "\n".join(lines) + + +def format_interval(secs: float) -> str: + """Human-readable duration (e.g. ``'5m'``, ``'1h30m'``).""" + if secs < 60: + return f"{secs:.0f}s" + if secs < 3600: + return f"{secs / 60:.0f}m" + hours = int(secs // 3600) + mins = int((secs % 3600) // 60) + if mins: + return f"{hours}h{mins}m" + return f"{hours}h" + + +# --------------------------------------------------------------------------- +# WatchRunner — server-level daemon thread +# --------------------------------------------------------------------------- + + +class WatchRunner: + """Polls the database for due watches and dispatches results. + + Runs as a daemon thread in the server process, analogous to + ``TaskScheduler`` in the console. + """ + + def __init__( + self, + storage: Any, + node_id: str, + *, + check_interval: float = 15.0, + tool_timeout: float = 30.0, + restore_fn: Callable[[str], Callable[[str], None] | None] | None = None, + ) -> None: + self._storage = storage + self._node_id = node_id + self._check_interval = check_interval + self._tool_timeout = tool_timeout + self._restore_fn = restore_fn + + self._dispatch_fns: dict[str, Callable[[str], None]] = {} + self._dispatch_lock = threading.Lock() + + self._stop_event = threading.Event() + self._thread: threading.Thread | None = None + + # -- Lifecycle ----------------------------------------------------------- + + def start(self) -> None: + if self._thread is not None: + return + self._stop_event.clear() + self._thread = threading.Thread(target=self._run, daemon=True, name="watch-runner") + self._thread.start() + log.info("watch_runner.started", extra={"node_id": self._node_id}) + + def stop(self) -> None: + self._stop_event.set() + if self._thread is not None: + self._thread.join(timeout=self._check_interval + 5) + self._thread = None + log.info("watch_runner.stopped") + + # -- Dispatch function registry ------------------------------------------ + + def set_dispatch_fn(self, ws_id: str, fn: Callable[[str], None]) -> None: + with self._dispatch_lock: + self._dispatch_fns[ws_id] = fn + + def remove_dispatch_fn(self, ws_id: str) -> None: + with self._dispatch_lock: + self._dispatch_fns.pop(ws_id, None) + + # -- Main loop ----------------------------------------------------------- + + def _run(self) -> None: + while not self._stop_event.is_set(): + try: + self._tick() + except Exception: + log.exception("watch_runner.tick_error") + self._stop_event.wait(self._check_interval) + + def _tick(self) -> None: + if self._storage is None: + return + now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + due = self._storage.list_due_watches(now) + for watch_row in due: + if self._stop_event.is_set(): + break + # Only poll watches owned by this node + row_node = watch_row.get("node_id", "") + if row_node and row_node != self._node_id: + continue + try: + self._poll_watch(watch_row) + except Exception: + log.exception( + "watch_runner.poll_error", + extra={"watch_id": watch_row.get("watch_id")}, + ) + + def _poll_watch(self, watch_row: dict[str, Any]) -> None: + watch_id = watch_row["watch_id"] + ws_id = watch_row["ws_id"] + command = watch_row["command"] + stop_on = watch_row.get("stop_on") + max_polls = watch_row.get("max_polls", DEFAULT_MAX_POLLS) + poll_count = watch_row.get("poll_count", 0) + 1 + prev_output = watch_row.get("last_output") + created = watch_row.get("created", "") + + # Safety check + blocked = is_command_blocked(command) + if blocked: + log.warning( + "watch_runner.blocked_command", extra={"watch_id": watch_id, "reason": blocked} + ) + self._deactivate_watch(watch_id) + return + + # Run command + output, exit_code = self._run_command(sanitize_command(command)) + + # Truncate to avoid unbounded storage / context window usage + if len(output) > MAX_OUTPUT_SIZE: + output = output[:MAX_OUTPUT_SIZE] + f"\n[truncated at {MAX_OUTPUT_SIZE} bytes]" + + # Evaluate condition + fired, reason = evaluate_condition(stop_on, output, exit_code, prev_output) + + # Treat condition evaluation errors as terminal — don't silently + # loop until max_polls while the user/model never sees the problem. + if not fired and reason.startswith("condition error:"): + fired = True + + # Check max polls + is_final = fired or poll_count >= max_polls + if not fired and poll_count >= max_polls: + reason = "max polls reached" + is_final = True + + now = datetime.now(UTC) + now_str = now.strftime("%Y-%m-%dT%H:%M:%S") + + # Update DB + update_fields: dict[str, Any] = { + "poll_count": poll_count, + "last_output": output, + "last_exit_code": exit_code, + "last_poll": now_str, + } + if is_final: + update_fields["active"] = False + update_fields["next_poll"] = "" + else: + next_poll = now + timedelta(seconds=watch_row["interval_secs"]) + update_fields["next_poll"] = next_poll.strftime("%Y-%m-%dT%H:%M:%S") + self._storage.update_watch(watch_id, **update_fields) + + # Dispatch result if condition fired or final + if fired or is_final: + # Compute elapsed from created time + elapsed_secs = 0.0 + if created: + try: + created_dt = datetime.fromisoformat(created).replace(tzinfo=UTC) + elapsed_secs = (now - created_dt).total_seconds() + except (ValueError, TypeError): + pass + + message = format_watch_message( + name=watch_row["name"], + command=command, + output=output, + poll_count=poll_count, + max_polls=max_polls, + elapsed_secs=elapsed_secs, + stop_on=stop_on, + is_final=is_final, + reason=reason, + ) + self._dispatch_result(ws_id, message) + + log.debug( + "watch_runner.polled", + extra={ + "watch_id": watch_id, + "poll_count": poll_count, + "fired": fired, + "is_final": is_final, + }, + ) + + def _run_command(self, command: str) -> tuple[str, int]: + """Run a shell command and return (stdout, exit_code).""" + try: + proc = subprocess.run( + command, + shell=True, + capture_output=True, + text=True, + timeout=self._tool_timeout, + start_new_session=True, + ) + output = proc.stdout + if proc.stderr: + output = output + "\n[stderr]\n" + proc.stderr if output else proc.stderr + return output, proc.returncode + except subprocess.TimeoutExpired: + return f"[command timed out after {self._tool_timeout}s]", -1 + except Exception as exc: + return f"[command failed: {exc}]", -1 + + def _dispatch_result(self, ws_id: str, message: str) -> None: + """Deliver a watch result to the owning workstream.""" + with self._dispatch_lock: + fn = self._dispatch_fns.get(ws_id) + + if fn is not None: + try: + fn(message) + return + except Exception: + log.exception("watch_runner.dispatch_error", extra={"ws_id": ws_id}) + + # Workstream may be evicted — try to restore + if self._restore_fn is not None: + try: + restored_fn = self._restore_fn(ws_id) + if restored_fn is not None: + restored_fn(message) + return + except Exception: + log.exception("watch_runner.restore_error", extra={"ws_id": ws_id}) + + log.warning( + "watch_runner.dispatch_failed", + extra={"ws_id": ws_id, "reason": "no dispatch function and restore failed"}, + ) + + def _deactivate_watch(self, watch_id: str) -> None: + self._storage.update_watch(watch_id, active=False, next_poll="") diff --git a/turnstone/server.py b/turnstone/server.py index a739d95f..b3840461 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -699,6 +699,35 @@ async def metrics_endpoint(request: Request) -> Response: return Response(content, media_type="text/plain; version=0.0.4; charset=utf-8") +def _make_watch_dispatch(ws: Workstream, session: ChatSession, ui: Any) -> Any: + """Create a dispatch function for watch results on a workstream. + + Handles both idle (start worker thread) and busy (enqueue for IDLE drain) + cases. Mirrors the ``send_message`` worker-thread pattern. + """ + pending = session._watch_pending + + def dispatch(msg: str) -> None: + if ws.worker_thread and ws.worker_thread.is_alive(): + # Workstream is busy — queue for drain at IDLE (Path A) + pending.put({"message": msg}) + return + + # Workstream is idle — start a worker thread (Path B) + def run() -> None: + try: + session.send(msg) + except Exception as exc: + if ui: + ui.on_error(f"Watch error: {exc}") + + t = threading.Thread(target=run, daemon=True) + ws.worker_thread = t + t.start() + + return dispatch + + async def send_message(request: Request) -> JSONResponse: """POST /v1/api/send — send a user message to the workstream.""" from turnstone.core.web_helpers import read_json_or_400 @@ -844,6 +873,12 @@ async def create_workstream(request: Request) -> JSONResponse: assert isinstance(ws.ui, WebUI) if skip or body.get("auto_approve", False): ws.ui.auto_approve = True + # Register watch runner for this workstream + runner = getattr(request.app.state, "watch_runner", None) + if runner and ws.session: + ws.session.set_watch_runner( + runner, dispatch_fn=_make_watch_dispatch(ws, ws.session, ws.ui) + ) # Emit eviction event if a workstream was evicted to make room evicted = mgr.last_evicted if evicted is not None: @@ -902,6 +937,42 @@ async def close_workstream(request: Request) -> JSONResponse: return JSONResponse({"error": "Cannot close last workstream"}, status_code=400) +async def list_watches(request: Request) -> JSONResponse: + """GET /v1/api/watches — list active watches, optionally filtered by ws_id.""" + from turnstone.core.storage._registry import get_storage + + storage = get_storage() + if not storage: + return JSONResponse({"watches": []}) + ws_id = request.query_params.get("ws_id") + if ws_id: + watches = storage.list_watches_for_ws(ws_id) + else: + node_id = getattr(request.app.state, "node_id", "") + watches = storage.list_watches_for_node(node_id) if node_id else [] + return JSONResponse({"watches": watches}) + + +async def cancel_watch(request: Request) -> JSONResponse: + """POST /v1/api/watches/{watch_id}/cancel — cancel an active watch.""" + from turnstone.core.storage._registry import get_storage + + watch_id = request.path_params["watch_id"] + storage = get_storage() + if not storage: + return JSONResponse({"error": "Storage unavailable"}, status_code=500) + watch = storage.get_watch(watch_id) + if not watch: + return JSONResponse({"error": "Watch not found"}, status_code=404) + # Verify node ownership in multi-node deployments + node_id = getattr(request.app.state, "node_id", "") + watch_node = watch.get("node_id", "") + if watch_node and node_id and watch_node != node_id: + return JSONResponse({"error": "Watch belongs to another node"}, status_code=403) + storage.update_watch(watch_id, active=False, next_poll="") + return JSONResponse({"status": "ok", "watch_id": watch_id}) + + async def auth_login(request: Request) -> Response: """POST /v1/api/auth/login — authenticate and return JWT.""" from turnstone.core.auth import handle_auth_login @@ -1003,8 +1074,13 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]: daemon=True, ) cleanup.start() + # Start watch runner (periodic command polling) + if app.state.watch_runner: + app.state.watch_runner.start() yield # Shutdown + if app.state.watch_runner: + app.state.watch_runner.stop() if app.state.health_monitor: app.state.health_monitor.stop() if app.state.mcp_client: @@ -1054,6 +1130,7 @@ def create_app( idle_timeout: int = 0, node_id: str = "", cors_origins: list[str] | None = None, + watch_runner: Any = None, ) -> Starlette: """Create and configure the Starlette ASGI application.""" _spec = build_server_spec() @@ -1077,6 +1154,8 @@ def create_app( Route("/api/command", command, methods=["POST"]), Route("/api/workstreams/new", create_workstream, methods=["POST"]), Route("/api/workstreams/close", close_workstream, methods=["POST"]), + Route("/api/watches", list_watches), + Route("/api/watches/{watch_id}/cancel", cancel_watch, methods=["POST"]), Route("/api/auth/login", auth_login, methods=["POST"]), Route("/api/auth/logout", auth_logout, methods=["POST"]), Route("/api/auth/status", auth_status), @@ -1107,6 +1186,7 @@ def create_app( app.state.registry = registry app.state.idle_timeout = idle_timeout app.state.node_id = node_id + app.state.watch_runner = watch_runner from turnstone.core.auth import LoginRateLimiter @@ -1492,11 +1572,47 @@ def main() -> None: tool_search_max_results=args.tool_search_max_results, ) - # Create workstream manager and initial workstream + # Create WatchRunner (periodic command polling, server-level) + from turnstone.core.storage import get_storage as _get_storage + from turnstone.core.watch import WatchRunner + + # Create workstream manager first (watch restore_fn captures it) manager = WorkstreamManager( session_factory, max_workstreams=args.max_workstreams, node_id=_node_id ) WebUI._workstream_mgr = manager + + def _watch_restore_fn(ws_id: str) -> Any: + """Restore an evicted workstream so a watch can deliver results. + + Returns a callable that starts a worker thread to send() the watch + result. Unlike the normal dispatch path (which enqueues for IDLE + drain), the restored workstream has no active send() loop, so we + must start a worker thread directly — same pattern as send_message(). + """ + try: + ws = manager.create( + ui_factory=lambda wid: WebUI(ws_id=wid), + ) + # Restored workstreams run unattended — auto-approve tool calls + # to avoid blocking forever on approval with no connected user. + if isinstance(ws.ui, WebUI): + ws.ui.auto_approve = True + if ws.session: + ws.session.resume(ws_id) + dispatch_fn = _make_watch_dispatch(ws, ws.session, ws.ui) + ws.session.set_watch_runner(_watch_runner, dispatch_fn=dispatch_fn) + return dispatch_fn + except RuntimeError: + log.warning("watch_restore: cannot restore ws %s (all slots active)", ws_id) + return None + + _watch_runner = WatchRunner( + storage=_get_storage(), + node_id=_node_id, + tool_timeout=args.tool_timeout, + restore_fn=_watch_restore_fn, + ) ws = manager.create( name="default", ui_factory=lambda wid: WebUI(ws_id=wid), @@ -1507,6 +1623,9 @@ def main() -> None: # Handle --resume assert ws.session is not None + ws.session.set_watch_runner( + _watch_runner, dispatch_fn=_make_watch_dispatch(ws, ws.session, ws.ui) + ) if args.resume: from turnstone.core.memory import resolve_workstream @@ -1552,6 +1671,7 @@ def main() -> None: idle_timeout=args.workstream_idle_timeout, node_id=_node_id, cors_origins=cors_origins, + watch_runner=_watch_runner, ) log.info("Server starting on http://%s:%s", args.host, args.port) diff --git a/turnstone/tools/watch.json b/turnstone/tools/watch.json new file mode 100644 index 00000000..3b69bb77 --- /dev/null +++ b/turnstone/tools/watch.json @@ -0,0 +1,36 @@ +{ + "name": "watch", + "description": "Set up periodic polling of a shell command within this workstream. Actions: 'create' starts a new watch, 'list' shows active watches, 'cancel' stops a watch. Watch results are injected into the conversation when the stop condition is met or output changes. Use for monitoring CI/CD, PR status, deployments, file changes, etc.", + "parameters": { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["create", "list", "cancel"], + "description": "Action to perform." + }, + "command": { + "type": "string", + "description": "Shell command to poll (required for 'create')." + }, + "poll_every": { + "type": "string", + "description": "Poll interval as duration (e.g., '30s', '5m', '1h'). Default: '5m'." + }, + "stop_on": { + "type": "string", + "description": "Python expression evaluated after each poll. Variables: output (str), data (parsed JSON or None), exit_code (int), prev_output (str|None), changed (bool). Truthy result fires the watch and auto-cancels. Omit for change-detection mode (first poll establishes a baseline, subsequent polls fire when output differs). Examples: 'data[\"state\"] == \"MERGED\"', '\"error\" in output', 'exit_code != 0'." + }, + "name": { + "type": "string", + "description": "Human-readable name (e.g., 'pr-review'). Required for 'create', used as identifier for 'cancel'." + }, + "max_polls": { + "type": "integer", + "description": "Max poll cycles before auto-cancel. Default: 100." + } + }, + "required": ["action"] + }, + "primary_key": "command" +}