Feat/coordinator phase2 audit (#369)

* feat(coordinator): audit middleware on routing proxy — phase 2

Adds per-tool-call audit attribution to the multi-node routing proxy
handlers so coordinator → server hops land observable rows in
``audit_events``.  Phase 1 preserved the ``src="coordinator"`` claim
through ``_proxy_auth_headers``'s upstream re-mint; this commit
closes the recording side.  Was the last real security gap from
phase 1 — an enterprise deployment with ``admin.coordinator``
granted got only the three console-side
``coordinator.{create,close,cancel}`` rows; per-tool-call
attribution was missing.

## Action-naming scheme

  route.workstream.create   POST /v1/api/route/workstreams/new
  route.workstream.send     POST /v1/api/route/send
  route.workstream.close    POST /v1/api/route/workstreams/close
  route.workstream.delete   POST /v1/api/route/workstreams/delete
  route.approve             POST /v1/api/route/approve
  route.cancel              POST /v1/api/route/cancel
  route.command             POST /v1/api/route/command
  route.plan                POST /v1/api/route/plan

Action-name conventions documented in ``turnstone/core/audit.py``
module docstring alongside the existing namespaces — the docstring
is now ``<resource>.<verb>`` shaped (non-exhaustive) rather than
trying to enumerate every prefix.

## Recording rules

- ``record_audit()`` fires only on a 2xx upstream response.  4xx/5xx
  are observable via ``_record_route``'s metrics path; doubling the
  audit-events table size for failure rows would dilute signal
  without giving operators much extra value.
- ``detail`` JSON carries ``{src, node_id, coord_ws_id?}`` — ``src``
  lands verbatim from ``auth.token_source`` so non-coordinator
  origins (``"jwt"``, ``"console-proxy"``) also get attribution;
  ``coord_ws_id`` only appears when the inbound JWT carried it.
- Wrapped in ``try/except`` + ``log.debug("route.audit_failed", ...)``
  defence-in-depth.  ``record_audit`` itself is fire-and-forget;
  the outer try guards against a programmer error in the call site.

## Routing-proxy specifics

- ``route_create``: emits at the post-multipart/JSON convergence
  ``if resp.status_code == 200`` block.  Both branches set
  ``audit_ws_id`` correctly — multipart from the query-string ws_id,
  JSON from ``body["ws_id"]`` (post-503-retry) or ``body["resume_ws"]``.
- ``route_proxy``: emits the URL-method-mapped action.  ``ref`` is
  reassigned to ``new_ref`` after a successful 404→cache-refresh
  retry so audit attribution uses the retried node, not the failed
  first node.
- ``route_workstream_delete``: emits on 2xx using the ws_id from
  the request body.
- ``route_attachment_proxy``: out of scope (upstream attachment
  endpoints emit their own ``workstream.attachment.*`` rows;
  auditing here would double-count).

## Tests

16 new tests in ``tests/test_route_proxy_audit.py`` covering:
- Coordinator-origin emission with full detail payload.
- 502 / 400 / 503-retry-final-node-id paths.
- Parametrised method→action mapping for the 6 ``route_proxy`` URLs.
- Plain-JWT origin (no ``coord_ws_id`` in detail).
- Delete handler 2xx + 502.
- Audit-storage exception swallowed (proxied response unchanged).
- ``auth_storage`` absent → no-op (existing route-handler tests
  unaffected).

Verification: ``ruff check`` clean, ``mypy turnstone`` clean
(156 source files), ``pytest -m "not live"`` 4087 passed,
3 deselected (live-backend), 0 failed.

* feat(coordinator): discovery tools and /open parity — list_nodes, list_skills, POST /coordinator/{ws_id}/open

Adds the read-side surface coordinators need to make informed
orchestration decisions plus an explicit rehydration endpoint
matching the server's ``POST /v1/api/workstreams/{ws_id}/open``.

## list_nodes (auto-approved)

``list_nodes(filters={key: value, ...})`` reads ``node_metadata`` via
``storage.filter_nodes_by_metadata`` + ``get_all_node_metadata`` —
one query each, no N+1.  Each row carries its full metadata dict so
the coordinator has both auto keys (``arch`` / ``cpu_count`` /
``fqdn`` / ``hostname`` / ``os`` / ``os_release`` / ``python``;
always present) and operator-supplied user keys (``capability`` /
``region`` / ``tenant`` / ``role``) without a second round-trip.
Tool description enumerates the auto keys explicitly so the model
knows what's always available vs deployment-specific.

Storage stores metadata values as JSON-encoded strings (the write
path in ``server.py`` / ``admin.py`` / ``console/server.py`` all go
through ``json.dumps``).  The client re-encodes filter values
before the stored-text comparison and decodes stored values before
returning them to the model — so ``{"capability": "gpu"}`` is the
natural form the model uses, not ``{"capability": "\"gpu\""}``.
Ints round-trip as ints.

Returns ``{nodes, truncated}``; ``truncated=True`` when the page
was full.

## list_skills (auto-approved)

``list_skills(category?, tag?, scan_status?, enabled_only?, limit?)``
surfaces the skill registry so coordinators can discover worker
profiles.  New storage protocol method ``list_skills_filtered(...)``
on both SQLite and PostgreSQL backends pushes filters into SQL.
``tag`` filter matches against the JSON-array ``tags`` column with
quote-bracketed substring (``%"tag"%``) — quote-safe against
``foo`` vs ``foobar`` collisions on both backends.

Returns ``{skills, truncated}`` with ``name`` / ``category`` /
``tags`` (decoded to list) / ``version`` / ``description`` /
``model`` / ``enabled`` / ``scan_status`` / ``activation`` — the
discovery projection, not the full row.

## POST /v1/api/coordinator/{ws_id}/open

Explicit rehydration endpoint.  Lazy ``GET`` rehydration works for
the UI; this gives SDK callers and operators a way to warm a
coordinator without browsing to it.  Same ownership / 404-on-
mismatch / correlation-id-masked error semantics as
``coordinator_detail``.  Returns ``{ws_id, name, already_loaded?}``.
Registered in ``turnstone/api/console_spec.py`` with a dedicated
``CoordinatorOpenResponse`` Pydantic model so the OpenAPI schema
matches the wire shape.

## Tests

- ``tests/test_storage_skills_filtered.py`` — 8 cases validated on
  BOTH SQLite and PostgreSQL backends (``pytest --storage-backend
  postgresql``).  Covers no-filter ordering, category exact-match,
  tag quote-safety (``"foo"`` matches ``["foo","bar"]`` but not
  ``["foobar"]``), scan_status, enabled_only, limit, AND semantics,
  empty result.
- ``tests/test_coordinator_client.py`` — 11 new cases covering
  node/skill shape decoding, JSON-encoded filter round-trip (the
  ``"gpu"`` vs ``'"gpu"'`` case), int filter encoding, truncation,
  no-match empty, no N+1 (``get_prompt_template`` /
  ``get_node_metadata`` call counts asserted zero).
- ``tests/test_coordinator_tools.py`` — 11 new cases for
  ``_prepare``/``_exec`` dispatch, filter type-drop, limit clamping
  (``limit=0`` falls back to 100, negatives clamp to 1),
  truncation-signal summary.
- ``tests/test_coordinator_endpoints.py`` — 8 new cases for
  ``/open``: ``already_loaded`` on in-memory hit, 404 on ownership
  mismatch, lazy rehydrate on miss, admin bypass, unknown ws_id,
  503 on ``coord_mgr`` unavailable, 500 with correlation-id mask on
  factory failure, 503 passthrough on ``ValueError``.
- ``tests/test_workstream_kind.py`` / ``test_tools_schema.py``
  updated to include ``list_nodes`` and ``list_skills`` in the
  disjoint-namespace regression guard and the tool-count check.

Verification: ``ruff check`` clean, ``mypy turnstone`` clean
(156 source files), ``pytest -m "not live"`` 4122 passed, 3
deselected (live-backend), 0 failed.  Postgres backend storage
tests green (``pytest --storage-backend postgresql
tests/test_storage_skills_filtered.py`` 8 passed).

* feat(coordinator): task_list tool — persistent planning state

Adds a coordinator-only ``task_list`` tool persisted on the
coordinator's own ``workstream_config`` row.  Gives coordinators a
scratch surface for work decomposition that survives restarts so the
UI can render planned-vs-done state once the tree view lands.

## Tool surface

``task_list(action, ...)`` with five actions:

- ``list``     auto-approved read.  Returns ``{tasks, truncated}``;
               truncated=True when the list exceeded the 200-row
               page cap.
- ``add``      needs approval.  ``title`` required; optional
               ``status`` and ``child_ws_id``.  Title clamped at
               200 chars.  Capacity cap at 500 tasks — hitting the
               cap is an explicit signal to prune done/blocked rows.
- ``update``   needs approval.  Mutate by ``task_id``; fields
               ``title`` / ``status`` / ``child_ws_id`` optional.
- ``remove``   needs approval.  Drop by ``task_id``.
- ``reorder``  needs approval.  Pass ``task_ids``; validated as an
               exact permutation of the current set (rejects
               partial, extra, or substituted ids — prevents silent
               task loss).

Status enum: ``pending`` / ``in_progress`` / ``done`` / ``blocked``.
``child_ws_id`` links a task to the child workstream spawned for it
(no enforcement; the coordinator owns the relationship).

## Persistence

Stored as a single JSON-envelope value on ``workstream_config`` —
``{"version": 1, "tasks": [...]}``.  No new table; the kanban v2
work will supersede this row via a format migration keyed on
``version``.  ``_save_task_list`` writes only the ``tasks`` key so
concurrent writers to other ``workstream_config`` keys (e.g. the
admin Settings UI updating ``reasoning_effort``) aren't clobbered
by a read-modify-write on the full row.

## Corrupt-envelope safety

A hand-edited or legacy config row that doesn't parse as the
expected shape logs a warning and returns an empty envelope from
``task_list_get``.  Mutators refuse to overwrite corrupt data —
they detect the sentinel and return a clear error so the operator
can inspect or clear the row rather than losing work silently.

## Concurrency

Per-(ws) ``threading.Lock`` cached on the client.  The worker
thread is single-threaded for tool execs so this is mostly
defence-in-depth against future maintenance-script call sites.
Cache never grows beyond one entry per coordinator session because
the scope guard short-circuits foreign ``ws_id`` before the lock
is acquired.

## Malformed-JSON recovery

``_prepare_tool`` fallback-1 regex-extract allowlist extended with
``action`` / ``status`` / ``task_id`` / ``title`` (alphabetized) so
slightly-malformed ``task_list`` calls get the same
self-correction behaviour as the other coordinator tools.

## Tests

- ``tests/test_coordinator_client.py`` — 15 new cases covering:
  fresh-envelope shape, add/get roundtrip, empty-title + invalid-
  status rejection, 200-char title clamp, update by id + missing
  id, remove semantics, reorder permutation validation (partial +
  extra + wrong id + valid), cross-ws scope violation, corrupt-
  JSON read recovery, corrupt-envelope write refusal (all four
  mutators), 500-task capacity cap, workstream_config key
  preservation across ``_save_task_list``.
- ``tests/test_coordinator_tools.py`` — 12 new cases covering the
  dispatch layer: list auto-approved, each mutating action needs
  approval, unknown-action / missing-required-arg errors, list
  returns tasks, page-cap at 200 with truncated signal, add
  dispatches to client, reorder surfaces permutation error,
  remove-not-found.
- ``tests/test_tools_schema.py`` / ``tests/test_workstream_kind.py``
  extend the tool-count + disjoint-namespace + primary-key
  regression guards with ``task_list``.

Verification: ``ruff check`` clean, ``mypy turnstone`` clean
(156 source files), ``pytest -m "not live"`` 4148 passed,
3 deselected (live-backend), 0 failed.

tests/test_coordinator_client.py

@@ -427,3 +427,390 @@ def test_list_children_truncated_signals_db_page_full(populated_storage):
     # always fills the page so truncated must fire.
     result = client.list_children("coord-1", limit=1)
     assert result["truncated"] is True
+
+
+# ---------------------------------------------------------------------------
+# list_nodes
+# ---------------------------------------------------------------------------
+
+
+def _set_meta(storage, node_id, entries):
+    """Write node metadata the way production writers do — JSON-encoded values.
+
+    ``server.py``, ``admin.py``, and ``console/server.py`` all call
+    ``set_node_metadata[_bulk]`` with ``json.dumps(value)``.  Tests have
+    to use the same encoding so coordinator filter semantics are
+    validated against realistic data.
+    """
+    storage.set_node_metadata_bulk(
+        node_id,
+        [(k, json.dumps(v), src) for (k, v, src) in entries],
+    )
+
+
+@pytest.fixture
+def storage_with_nodes(tmp_path):
+    st = SQLiteBackend(str(tmp_path / "nodes.db"))
+    _set_meta(
+        st,
+        "node-a",
+        [
+            ("arch", "x86_64", "auto"),
+            ("cpu_count", 4, "auto"),
+            ("region", "us-east", "user"),
+        ],
+    )
+    _set_meta(
+        st,
+        "node-b",
+        [
+            ("arch", "x86_64", "auto"),
+            ("cpu_count", 16, "auto"),
+            ("region", "us-west", "user"),
+            ("capability", "gpu", "user"),
+        ],
+    )
+    _set_meta(
+        st,
+        "node-c",
+        [("arch", "arm64", "auto"), ("cpu_count", 8, "auto")],
+    )
+    return st
+
+
+def test_list_nodes_no_filters_returns_all_rows_decoded(storage_with_nodes):
+    client = _make_read_client(storage_with_nodes)
+    result = client.list_nodes()
+    assert set(result.keys()) == {"nodes", "truncated"}
+    node_ids = {n["node_id"] for n in result["nodes"]}
+    assert node_ids == {"node-a", "node-b", "node-c"}
+    assert result["truncated"] is False
+    # Values round-trip through json.loads — model sees natural types,
+    # not the raw stored JSON text.
+    node_b = next(n for n in result["nodes"] if n["node_id"] == "node-b")
+    assert node_b["metadata"]["arch"] == {"value": "x86_64", "source": "auto"}
+    assert node_b["metadata"]["cpu_count"] == {"value": 16, "source": "auto"}
+    assert node_b["metadata"]["capability"] == {"value": "gpu", "source": "user"}
+
+
+def test_list_nodes_filter_uses_natural_value_not_quoted(storage_with_nodes, monkeypatch):
+    """Model passes ``{"capability": "gpu"}`` — client re-encodes to
+    ``'"gpu"'`` before filter_nodes_by_metadata so the stored text
+    matches.  Also asserts the filtered path fetches metadata only for
+    the paginated slice (bounded at page_size) rather than the whole
+    cluster — no wide ``get_all_node_metadata`` scan on a narrow filter.
+    """
+    per_node_calls: list[str] = []
+    real = storage_with_nodes.get_node_metadata
+
+    def _spy(nid):  # type: ignore[no-untyped-def]
+        per_node_calls.append(nid)
+        return real(nid)
+
+    all_meta_calls: list[int] = []
+    real_all = storage_with_nodes.get_all_node_metadata
+
+    def _spy_all():  # type: ignore[no-untyped-def]
+        all_meta_calls.append(1)
+        return real_all()
+
+    monkeypatch.setattr(storage_with_nodes, "get_node_metadata", _spy)
+    monkeypatch.setattr(storage_with_nodes, "get_all_node_metadata", _spy_all)
+
+    client = _make_read_client(storage_with_nodes)
+    result = client.list_nodes(filters={"capability": "gpu"})
+    assert {n["node_id"] for n in result["nodes"]} == {"node-b"}
+    # Filtered path: no wide scan; per-node lookups bounded to the
+    # matching page (1 row matched the filter).
+    assert all_meta_calls == []
+    assert per_node_calls == ["node-b"]
+
+
+def test_list_nodes_filter_accepts_int_and_encodes_correctly(storage_with_nodes):
+    """Model passes ``{"cpu_count": 4}`` — int encoded to ``"4"``; match."""
+    client = _make_read_client(storage_with_nodes)
+    result = client.list_nodes(filters={"cpu_count": 4})
+    assert {n["node_id"] for n in result["nodes"]} == {"node-a"}
+
+
+def test_list_nodes_int_and_string_filters_are_distinct(storage_with_nodes):
+    """The JSON schema for ``filters`` accepts primitives (string, integer,
+    number, boolean); stringified ints compare as strings, not as ints.
+    The tool description documents this as ``JSON-equal compare``.
+    """
+    client = _make_read_client(storage_with_nodes)
+    # Int filter against int-stored value matches.
+    assert {n["node_id"] for n in client.list_nodes(filters={"cpu_count": 4})["nodes"]} == {
+        "node-a"
+    }
+    # String filter against int-stored value is a distinct comparison and
+    # returns zero rows — ``"4"`` JSON-encodes to ``'"4"'`` but the stored
+    # row is ``'4'``.  Documented in the tool description.
+    assert client.list_nodes(filters={"cpu_count": "4"})["nodes"] == []
+
+
+def test_list_nodes_truncation_signal(storage_with_nodes):
+    client = _make_read_client(storage_with_nodes)
+    result = client.list_nodes(limit=2)
+    assert len(result["nodes"]) == 2
+    assert result["truncated"] is True
+
+
+def test_list_nodes_empty_on_no_matching_filters(storage_with_nodes):
+    client = _make_read_client(storage_with_nodes)
+    result = client.list_nodes(filters={"region": "nowhere"})
+    assert result["nodes"] == []
+    assert result["truncated"] is False
+
+
+# ---------------------------------------------------------------------------
+# list_skills
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def storage_with_skills(tmp_path):
+    st = SQLiteBackend(str(tmp_path / "skills.db"))
+    st.create_prompt_template(
+        template_id="s1",
+        name="alpha",
+        category="ops",
+        content="",
+        variables="[]",
+        is_default=False,
+        org_id="",
+        created_by="test",
+        tags='["gpu", "fast"]',
+    )
+    st.create_prompt_template(
+        template_id="s2",
+        name="beta",
+        category="engineering",
+        content="",
+        variables="[]",
+        is_default=False,
+        org_id="",
+        created_by="test",
+        tags='["slow"]',
+    )
+    st.create_prompt_template(
+        template_id="s3",
+        name="gamma",
+        category="engineering",
+        content="",
+        variables="[]",
+        is_default=False,
+        org_id="",
+        created_by="test",
+        tags="[]",
+        enabled=False,
+    )
+    return st
+
+
+def test_list_skills_returns_shape(storage_with_skills):
+    client = _make_read_client(storage_with_skills)
+    result = client.list_skills()
+    assert set(result.keys()) == {"skills", "truncated"}
+    names = {s["name"] for s in result["skills"]}
+    assert names == {"alpha", "beta", "gamma"}
+    # Tags decoded to a list, not a string.
+    alpha = next(s for s in result["skills"] if s["name"] == "alpha")
+    assert alpha["tags"] == ["gpu", "fast"]
+    # Discovery projection only — not full row.
+    assert "content" not in alpha
+
+
+def test_list_skills_pushes_filters_to_storage_no_per_row_lookups(storage_with_skills, monkeypatch):
+    called = []
+    real_get = storage_with_skills.get_prompt_template
+
+    def _spy(tid):  # type: ignore[no-untyped-def]
+        called.append(tid)
+        return real_get(tid)
+
+    monkeypatch.setattr(storage_with_skills, "get_prompt_template", _spy)
+
+    client = _make_read_client(storage_with_skills)
+    result = client.list_skills(tag="gpu")
+    assert {s["name"] for s in result["skills"]} == {"alpha"}
+    assert called == []  # no N+1
+
+
+def test_list_skills_enabled_only(storage_with_skills):
+    client = _make_read_client(storage_with_skills)
+    result = client.list_skills(enabled_only=True)
+    names = {s["name"] for s in result["skills"]}
+    assert names == {"alpha", "beta"}  # gamma is disabled
+
+
+def test_list_skills_truncation_signal(storage_with_skills):
+    client = _make_read_client(storage_with_skills)
+    result = client.list_skills(limit=2)
+    assert len(result["skills"]) == 2
+    assert result["truncated"] is True
+
+
+# ---------------------------------------------------------------------------
+# task_list
+# ---------------------------------------------------------------------------
+
+
+def _task_client(tmp_path) -> CoordinatorClient:
+    st = SQLiteBackend(str(tmp_path / "tasks.db"))
+    st.register_workstream("coord-1", kind="coordinator", user_id="user-1")
+    return _make_read_client(st)
+
+
+def test_task_list_get_empty_envelope_on_fresh_ws(tmp_path):
+    client = _task_client(tmp_path)
+    env = client.task_list_get("coord-1")
+    assert env == {"version": 1, "tasks": []}
+
+
+def test_task_list_add_then_get_roundtrip(tmp_path):
+    client = _task_client(tmp_path)
+    task = client.task_list_add("coord-1", title="spawn worker")
+    assert task["title"] == "spawn worker"
+    assert task["status"] == "pending"
+    env = client.task_list_get("coord-1")
+    assert len(env["tasks"]) == 1
+    assert env["tasks"][0]["id"] == task["id"]
+
+
+def test_task_list_add_rejects_empty_title(tmp_path):
+    client = _task_client(tmp_path)
+    result = client.task_list_add("coord-1", title="   ")
+    assert "error" in result
+
+
+def test_task_list_add_rejects_invalid_status(tmp_path):
+    client = _task_client(tmp_path)
+    result = client.task_list_add("coord-1", title="x", status="nonsense")
+    assert "error" in result
+
+
+def test_task_list_add_clamps_title_to_200(tmp_path):
+    client = _task_client(tmp_path)
+    long_title = "a" * 500
+    task = client.task_list_add("coord-1", title=long_title)
+    assert len(task["title"]) == 200
+
+
+def test_task_list_update_by_id(tmp_path):
+    client = _task_client(tmp_path)
+    added = client.task_list_add("coord-1", title="plan")
+    updated = client.task_list_update(
+        "coord-1", task_id=added["id"], status="done", child_ws_id="ws-child"
+    )
+    assert updated["status"] == "done"
+    assert updated["child_ws_id"] == "ws-child"
+
+
+def test_task_list_update_missing_id(tmp_path):
+    client = _task_client(tmp_path)
+    result = client.task_list_update("coord-1", task_id="nope", status="done")
+    assert "error" in result
+
+
+def test_task_list_remove(tmp_path):
+    client = _task_client(tmp_path)
+    added = client.task_list_add("coord-1", title="plan")
+    first = client.task_list_remove("coord-1", task_id=added["id"])
+    assert first.get("ok") is True
+    assert first.get("task_id") == added["id"]
+    # Second remove of the same id returns a distinguishable not-found
+    # error (NOT a silent False that would mask a corrupt envelope).
+    second = client.task_list_remove("coord-1", task_id=added["id"])
+    assert "error" in second
+    assert "not found" in second["error"]
+    assert client.task_list_get("coord-1")["tasks"] == []
+
+
+def test_task_list_reorder_requires_permutation(tmp_path):
+    client = _task_client(tmp_path)
+    a = client.task_list_add("coord-1", title="a")
+    b = client.task_list_add("coord-1", title="b")
+    # Partial set — must reject.
+    bad = client.task_list_reorder("coord-1", task_ids=[a["id"]])
+    assert "error" in bad
+    # Wrong id — reject.
+    wrong = client.task_list_reorder("coord-1", task_ids=[a["id"], "ghost"])
+    assert "error" in wrong
+    # Valid permutation — accept.
+    ok = client.task_list_reorder("coord-1", task_ids=[b["id"], a["id"]])
+    assert ok.get("ok") is True
+    env = client.task_list_get("coord-1")
+    assert [t["id"] for t in env["tasks"]] == [b["id"], a["id"]]
+
+
+def test_task_list_cross_ws_scope_violation_is_noop(tmp_path):
+    client = _task_client(tmp_path)
+    # Client is bound to coord-1; anything else returns an empty envelope
+    # or an error without touching storage.
+    assert client.task_list_get("other-ws") == {"version": 1, "tasks": []}
+    res_add = client.task_list_add("other-ws", title="sneak")
+    assert "error" in res_add
+    res_remove = client.task_list_remove("other-ws", task_id="x")
+    assert "error" in res_remove
+    assert "scope violation" in res_remove["error"]
+
+
+def test_task_list_corrupt_json_returns_empty_envelope(tmp_path):
+    """A hand-edited / corrupt config row must not crash the tool."""
+    st = SQLiteBackend(str(tmp_path / "tasks.db"))
+    st.register_workstream("coord-1", kind="coordinator", user_id="user-1")
+    st.save_workstream_config("coord-1", {"tasks": "{not json"})
+    client = _make_read_client(st)
+    env = client.task_list_get("coord-1")
+    assert env == {"version": 1, "tasks": []}
+
+
+def test_task_list_mutations_refuse_corrupt_envelope(tmp_path):
+    """When the envelope is corrupt on disk, mutators must error out
+    (rather than silently overwrite — lost-data safety)."""
+    st = SQLiteBackend(str(tmp_path / "tasks.db"))
+    st.register_workstream("coord-1", kind="coordinator", user_id="user-1")
+    st.save_workstream_config("coord-1", {"tasks": "{not json"})
+    client = _make_read_client(st)
+    add_result = client.task_list_add("coord-1", title="new")
+    assert "error" in add_result
+    assert "corrupt" in add_result["error"]
+    # Also: the corrupt blob is preserved after the refused mutation.
+    assert st.load_workstream_config("coord-1").get("tasks") == "{not json"
+    update_result = client.task_list_update("coord-1", task_id="x", status="done")
+    assert "error" in update_result
+    reorder_result = client.task_list_reorder("coord-1", task_ids=[])
+    assert "error" in reorder_result
+    remove_result = client.task_list_remove("coord-1", task_id="x")
+    assert "error" in remove_result
+    assert "corrupt" in remove_result["error"]
+
+
+def test_task_list_add_enforces_capacity_cap(tmp_path, monkeypatch):
+    from turnstone.console import coordinator_client as cc_module
+
+    monkeypatch.setattr(cc_module, "_TASK_LIST_MAX", 3)
+    client = _task_client(tmp_path)
+    for i in range(3):
+        client.task_list_add("coord-1", title=f"t{i}")
+    overflow = client.task_list_add("coord-1", title="no-room")
+    assert "error" in overflow
+    assert "capacity" in overflow["error"]
+    # After a remove, add succeeds again.
+    env = client.task_list_get("coord-1")
+    client.task_list_remove("coord-1", task_id=env["tasks"][0]["id"])
+    added = client.task_list_add("coord-1", title="retry")
+    assert "error" not in added
+
+
+def test_task_list_save_preserves_other_workstream_config_keys(tmp_path):
+    """_save_task_list writes only the 'tasks' key so other keys survive."""
+    st = SQLiteBackend(str(tmp_path / "tasks.db"))
+    st.register_workstream("coord-1", kind="coordinator", user_id="user-1")
+    st.save_workstream_config("coord-1", {"reasoning_effort": "high"})
+    client = _make_read_client(st)
+    client.task_list_add("coord-1", title="plan")
+    config = st.load_workstream_config("coord-1")
+    assert config.get("reasoning_effort") == "high"
+    assert config.get("tasks")  # task_list wrote its key too

tests/test_coordinator_endpoints.py

@@ -28,6 +28,7 @@ from turnstone.console.server import (
     coordinator_detail,
     coordinator_history,
     coordinator_list,
+    coordinator_open,
     coordinator_send,
 )
 from turnstone.core.auth import AuthResult
@@ -135,6 +136,11 @@ def _make_client(
                 coordinator_history,
                 methods=["GET"],
             ),
+            Route(
+                "/v1/api/coordinator/{ws_id}/open",
+                coordinator_open,
+                methods=["POST"],
+            ),
             Route(
                 "/v1/api/coordinator/{ws_id}",
                 coordinator_detail,
@@ -434,3 +440,99 @@ def test_cancel_resolves_pending_approval(storage):
     resp = client.post(f"/v1/api/coordinator/{ws.id}/cancel", headers=_COORD_HEADERS)
     assert resp.status_code == 200
     assert ws.ui._approval_event.is_set()
+
+
+# ---------------------------------------------------------------------------
+# Open (explicit rehydration)
+# ---------------------------------------------------------------------------
+
+
+def test_open_returns_already_loaded_when_in_memory(storage):
+    mgr = _build_mgr(storage)
+    ws = mgr.create(user_id="user-1", name="live")
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post(f"/v1/api/coordinator/{ws.id}/open", headers=_COORD_HEADERS)
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["ws_id"] == ws.id
+    assert body.get("already_loaded") is True
+
+
+def test_open_returns_404_on_ownership_mismatch_in_memory(storage):
+    mgr = _build_mgr(storage)
+    ws = mgr.create(user_id="owner", name="theirs")
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post(
+        f"/v1/api/coordinator/{ws.id}/open",
+        headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
+    )
+    assert resp.status_code == 404  # not 403 — don't leak existence
+
+
+def test_open_rehydrates_when_not_in_memory(storage, monkeypatch):
+    mgr = _build_mgr(storage)
+    rehydrated = MagicMock()
+    rehydrated.id = "coord-rehy"
+    rehydrated.name = "rehydrated"
+    rehydrated.user_id = "user-1"
+    monkeypatch.setattr(mgr, "open", MagicMock(return_value=rehydrated))
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post("/v1/api/coordinator/coord-rehy/open", headers=_COORD_HEADERS)
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["ws_id"] == "coord-rehy"
+    assert body["name"] == "rehydrated"
+    assert "already_loaded" not in body
+    mgr.open.assert_called_once_with("coord-rehy", "user-1")
+
+
+def test_open_admin_uses_open_admin(storage, monkeypatch):
+    mgr = _build_mgr(storage)
+    rehydrated = MagicMock()
+    rehydrated.id = "coord-rehy"
+    rehydrated.name = "r"
+    rehydrated.user_id = "someone-else"
+    monkeypatch.setattr(mgr, "open_admin", MagicMock(return_value=rehydrated))
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post(
+        "/v1/api/coordinator/coord-rehy/open",
+        headers={
+            "X-Test-User": "admin-1",
+            "X-Test-Perms": "admin.coordinator,admin.users",
+        },
+    )
+    assert resp.status_code == 200
+    mgr.open_admin.assert_called_once_with("coord-rehy")
+
+
+def test_open_returns_404_when_unknown_ws_id(storage, monkeypatch):
+    mgr = _build_mgr(storage)
+    monkeypatch.setattr(mgr, "open", MagicMock(return_value=None))
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post("/v1/api/coordinator/nonexistent/open", headers=_COORD_HEADERS)
+    assert resp.status_code == 404
+
+
+def test_open_503_on_coord_mgr_unavailable(storage):
+    client = _make_client(storage, coord_mgr=None)
+    resp = client.post("/v1/api/coordinator/any-ws/open", headers=_COORD_HEADERS)
+    assert resp.status_code == 503
+
+
+def test_open_correlation_id_on_factory_failure(storage, monkeypatch):
+    mgr = _build_mgr(storage)
+    monkeypatch.setattr(mgr, "open", MagicMock(side_effect=RuntimeError("boom")))
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post("/v1/api/coordinator/bad-ws/open", headers=_COORD_HEADERS)
+    assert resp.status_code == 500
+    assert "correlation_id=" in resp.json()["error"]
+
+
+def test_open_503_when_open_raises_value_error(storage, monkeypatch):
+    """ValueError from the factory surfaces as 503 with the remediation text."""
+    mgr = _build_mgr(storage)
+    monkeypatch.setattr(mgr, "open", MagicMock(side_effect=ValueError("coord registry missing")))
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.post("/v1/api/coordinator/bad-ws/open", headers=_COORD_HEADERS)
+    assert resp.status_code == 503
+    assert "registry missing" in resp.json()["error"]

tests/test_coordinator_tools.py

@@ -124,6 +124,9 @@ def test_coordinator_session_uses_coordinator_tools(coord_session):
         "close_workstream",
         "delete_workstream",
         "list_workstreams",
+        "list_nodes",
+        "list_skills",
+        "task_list",
     }
     # Sub-agent tool sets are zeroed on coordinator sessions.
     assert sess._task_tools == []
@@ -408,6 +411,361 @@ def test_prepare_fails_cleanly_when_coord_client_missing(monkeypatch):
         ("close_workstream", {"ws_id": "x"}),
         ("delete_workstream", {"ws_id": "x"}),
         ("list_workstreams", {}),
+        ("list_nodes", {}),
+        ("list_skills", {}),
+        ("task_list", {"action": "list"}),
     ):
         item = sess._prepare_tool(_tc(tool, args))
         assert "error" in item, f"{tool} did not error on missing coord_client"
+
+
+# ---------------------------------------------------------------------------
+# list_nodes
+# ---------------------------------------------------------------------------
+
+
+def test_list_nodes_prepare_is_auto_approved(coord_session):
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(_tc("list_nodes", {}))
+    assert item["needs_approval"] is False
+    assert item["filters"] == {}
+    assert item["limit"] == 100
+
+
+def test_list_nodes_prepare_accepts_filters(coord_session):
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(
+        _tc("list_nodes", {"filters": {"arch": "x86_64", "capability": "gpu"}})
+    )
+    assert item["filters"] == {"arch": "x86_64", "capability": "gpu"}
+
+
+def test_list_nodes_prepare_drops_invalid_filter_types(coord_session):
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(
+        _tc(
+            "list_nodes",
+            {"filters": {"arch": "x86_64", "bad": {"nested": "dict"}, "": "empty-key"}},
+        )
+    )
+    # Nested dict values + empty keys are filtered out; string + primitive kept.
+    assert item["filters"] == {"arch": "x86_64"}
+
+
+def test_list_nodes_prepare_clamps_limit(coord_session):
+    sess, _coord, _ui = coord_session
+    over = sess._prepare_tool(_tc("list_nodes", {"limit": 9999}))
+    assert over["limit"] == 500
+    # limit == 0 falls back to the default (100), not 1 — consistent with
+    # the other coordinator list tools' ``int(args.get("limit") or 100)``.
+    zero = sess._prepare_tool(_tc("list_nodes", {"limit": 0}))
+    assert zero["limit"] == 100
+    neg = sess._prepare_tool(_tc("list_nodes", {"limit": -5}))
+    assert neg["limit"] == 1  # negative values clamp to 1
+
+
+def test_list_nodes_exec_dispatches_to_client(coord_session):
+    sess, coord, ui = coord_session
+    coord.list_nodes.return_value = {
+        "nodes": [{"node_id": "n1", "metadata": {"arch": {"value": "x86_64", "source": "auto"}}}],
+        "truncated": False,
+    }
+    item = sess._prepare_tool(_tc("list_nodes", {"filters": {"arch": "x86_64"}}))
+    call_id, output = sess._exec_list_nodes(item)
+    assert call_id == "call-1"
+    parsed = json.loads(output)
+    assert parsed["nodes"][0]["node_id"] == "n1"
+    assert parsed["truncated"] is False
+    coord.list_nodes.assert_called_once_with(filters={"arch": "x86_64"}, limit=100)
+
+
+def test_list_nodes_exec_surfaces_truncated_sentinel(coord_session):
+    sess, coord, ui = coord_session
+    coord.list_nodes.return_value = {"nodes": [], "truncated": True}
+    item = sess._prepare_tool(_tc("list_nodes", {}))
+    _, _ = sess._exec_list_nodes(item)
+    # Summary reported to UI carries the "truncated" hint.
+    assert any("truncated" in r[2] for r in ui.tool_results)
+
+
+# ---------------------------------------------------------------------------
+# list_skills
+# ---------------------------------------------------------------------------
+
+
+def test_list_skills_prepare_is_auto_approved(coord_session):
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(_tc("list_skills", {}))
+    assert item["needs_approval"] is False
+    assert item["category"] is None
+    assert item["tag"] is None
+    assert item["scan_status"] is None
+    assert item["enabled_only"] is False
+    assert item["limit"] == 100
+
+
+def test_list_skills_prepare_accepts_filters(coord_session):
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(
+        _tc(
+            "list_skills",
+            {"category": "ops", "tag": "gpu", "scan_status": "clean", "enabled_only": True},
+        )
+    )
+    assert item["category"] == "ops"
+    assert item["tag"] == "gpu"
+    assert item["scan_status"] == "clean"
+    assert item["enabled_only"] is True
+
+
+def test_list_skills_prepare_tolerates_non_string_filters(coord_session):
+    """A malformed model call with non-string filter values must NOT
+    raise AttributeError during ``.strip()`` — the prepare path should
+    coerce non-strings to ``None`` and proceed."""
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(
+        _tc(
+            "list_skills",
+            {"category": 42, "tag": ["not", "a", "string"], "scan_status": {"bad": 1}},
+        )
+    )
+    assert "error" not in item
+    assert item["category"] is None
+    assert item["tag"] is None
+    assert item["scan_status"] is None
+
+
+def test_list_skills_prepare_parses_enabled_only_string_forms(coord_session):
+    """``bool("false")`` is True (non-empty string).  The prepare path
+    must interpret common string forms the way the model would expect."""
+    sess, _coord, _ui = coord_session
+    for raw, expected in (
+        ("true", True),
+        ("True", True),
+        ("1", True),
+        ("false", False),
+        ("False", False),
+        ("0", False),
+        ("", False),
+        (True, True),
+        (False, False),
+    ):
+        item = sess._prepare_tool(_tc("list_skills", {"enabled_only": raw}))
+        assert item.get("enabled_only") is expected, (
+            f"enabled_only={raw!r} → {item.get('enabled_only')!r}, expected {expected!r}"
+        )
+
+
+def test_list_skills_exec_dispatches_to_client(coord_session):
+    sess, coord, ui = coord_session
+    coord.list_skills.return_value = {
+        "skills": [{"name": "alpha", "tags": ["gpu"]}],
+        "truncated": False,
+    }
+    item = sess._prepare_tool(_tc("list_skills", {"category": "ops", "tag": "gpu"}))
+    call_id, output = sess._exec_list_skills(item)
+    assert call_id == "call-1"
+    parsed = json.loads(output)
+    assert parsed["skills"][0]["name"] == "alpha"
+    coord.list_skills.assert_called_once_with(
+        category="ops",
+        tag="gpu",
+        scan_status=None,
+        enabled_only=False,
+        limit=100,
+    )
+
+
+def test_list_skills_exec_surfaces_truncated_sentinel(coord_session):
+    sess, coord, ui = coord_session
+    coord.list_skills.return_value = {"skills": [], "truncated": True}
+    item = sess._prepare_tool(_tc("list_skills", {}))
+    _, _ = sess._exec_list_skills(item)
+    assert any("truncated" in r[2] for r in ui.tool_results)
+
+
+# ---------------------------------------------------------------------------
+# task_list
+# ---------------------------------------------------------------------------
+
+
+def test_task_list_list_is_auto_approved(coord_session):
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(_tc("task_list", {"action": "list"}))
+    assert item["needs_approval"] is False
+    assert item["action"] == "list"
+
+
+def test_task_list_bare_string_fallback_uses_action_primary_key(coord_session):
+    """A model that emits an unquoted ``list`` as the arguments blob
+    lands on the ``primary_key=action`` fallback and recovers.  Before
+    the fix primary_key was ``title`` so the fallback produced
+    ``{"title": "list"}`` and hit the required-action rejection."""
+    sess, _coord, _ui = coord_session
+    call = {
+        "id": "c1",
+        "type": "function",
+        "function": {"name": "task_list", "arguments": "list"},
+    }
+    item = sess._prepare_tool(call)
+    assert "error" not in item
+    assert item["action"] == "list"
+
+
+def test_task_list_mutating_actions_need_approval(coord_session):
+    sess, _coord, _ui = coord_session
+    add_item = sess._prepare_tool(_tc("task_list", {"action": "add", "title": "plan"}))
+    assert add_item["needs_approval"] is True
+    update_item = sess._prepare_tool(
+        _tc("task_list", {"action": "update", "task_id": "tsk_1", "status": "done"})
+    )
+    assert update_item["needs_approval"] is True
+    remove_item = sess._prepare_tool(_tc("task_list", {"action": "remove", "task_id": "tsk_1"}))
+    assert remove_item["needs_approval"] is True
+    reorder_item = sess._prepare_tool(
+        _tc("task_list", {"action": "reorder", "task_ids": ["tsk_1"]})
+    )
+    assert reorder_item["needs_approval"] is True
+
+
+def test_task_list_unknown_action_errors(coord_session):
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(_tc("task_list", {"action": "wat"}))
+    assert "error" in item
+
+
+def test_task_list_non_string_action_errors_cleanly(coord_session):
+    """A malformed ``action=42`` must NOT raise AttributeError during
+    ``.strip().lower()`` — coerce to the empty string and fall through
+    to the enum-check error."""
+    sess, _coord, _ui = coord_session
+    for bad_action in (42, None, ["list"], {"a": 1}, True):
+        item = sess._prepare_tool(_tc("task_list", {"action": bad_action}))
+        assert "error" in item, f"action={bad_action!r} did not produce a clean error"
+
+
+def test_task_list_add_rejects_non_string_title_and_status(coord_session):
+    """Add branch: ``title=42`` / ``status=0`` must NOT raise
+    AttributeError during ``.strip()``; produce a clean error item."""
+    sess, _coord, _ui = coord_session
+    for bad in ({"action": "add", "title": 42}, {"action": "add", "title": "ok", "status": 0}):
+        item = sess._prepare_tool(_tc("task_list", bad))
+        assert "error" in item, f"args={bad!r} did not produce a clean error"
+
+
+def test_task_list_remove_non_string_task_id_errors_cleanly(coord_session):
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(_tc("task_list", {"action": "remove", "task_id": 42}))
+    assert "error" in item
+
+
+def test_task_list_add_requires_title(coord_session):
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(_tc("task_list", {"action": "add", "title": ""}))
+    assert "error" in item
+
+
+def test_task_list_update_requires_task_id(coord_session):
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(_tc("task_list", {"action": "update", "status": "done"}))
+    assert "error" in item
+
+
+def test_task_list_update_rejects_non_string_field_values(coord_session):
+    """Preview must not diverge from execute: reject non-string field
+    values at prepare time rather than silently coercing to None."""
+    sess, _coord, _ui = coord_session
+    for field in ("title", "status", "child_ws_id"):
+        item = sess._prepare_tool(
+            _tc("task_list", {"action": "update", "task_id": "t1", field: 42})
+        )
+        assert "error" in item, f"update with non-string {field} should error"
+
+
+def test_task_list_update_requires_at_least_one_field(coord_session):
+    """update with only task_id is a no-op — reject to save an approval prompt."""
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(_tc("task_list", {"action": "update", "task_id": "t1"}))
+    assert "error" in item
+
+
+def test_task_list_remove_requires_task_id(coord_session):
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(_tc("task_list", {"action": "remove"}))
+    assert "error" in item
+
+
+def test_task_list_reorder_requires_list_of_strings(coord_session):
+    sess, _coord, _ui = coord_session
+    item = sess._prepare_tool(_tc("task_list", {"action": "reorder", "task_ids": [1, 2]}))
+    assert "error" in item
+
+
+def test_task_list_exec_list_returns_tasks(coord_session):
+    sess, coord, _ui = coord_session
+    coord.task_list_get.return_value = {
+        "version": 1,
+        "tasks": [{"id": "tsk_1", "title": "do", "status": "pending"}],
+    }
+    item = sess._prepare_tool(_tc("task_list", {"action": "list"}))
+    _, output = sess._exec_task_list(item)
+    parsed = json.loads(output)
+    assert parsed["tasks"][0]["id"] == "tsk_1"
+    assert parsed["truncated"] is False
+
+
+def test_task_list_exec_list_page_caps_at_200(coord_session):
+    sess, coord, _ui = coord_session
+    coord.task_list_get.return_value = {
+        "version": 1,
+        "tasks": [{"id": f"tsk_{i}", "title": "x", "status": "pending"} for i in range(250)],
+    }
+    item = sess._prepare_tool(_tc("task_list", {"action": "list"}))
+    _, output = sess._exec_task_list(item)
+    parsed = json.loads(output)
+    assert len(parsed["tasks"]) == 200
+    assert parsed["truncated"] is True
+
+
+def test_task_list_exec_add_dispatches(coord_session):
+    sess, coord, _ui = coord_session
+    coord.task_list_add.return_value = {"id": "tsk_new", "title": "plan"}
+    item = sess._prepare_tool(
+        _tc("task_list", {"action": "add", "title": "plan", "status": "pending"})
+    )
+    _, _ = sess._exec_task_list(item)
+    coord.task_list_add.assert_called_once_with(
+        sess._ws_id, title="plan", status="pending", child_ws_id=""
+    )
+
+
+def test_task_list_exec_reorder_surfaces_permutation_error(coord_session):
+    sess, coord, _ui = coord_session
+    coord.task_list_reorder.return_value = {"error": "task_ids must be a permutation..."}
+    item = sess._prepare_tool(_tc("task_list", {"action": "reorder", "task_ids": ["wrong"]}))
+    _, output = sess._exec_task_list(item)
+    parsed = json.loads(output)
+    assert "error" in parsed
+
+
+def test_task_list_exec_remove_passes_client_dict_through(coord_session):
+    """The client returns a dict; exec must pass it through without
+    synthesising a generic 'not found' message that would mask corrupt-
+    envelope errors from the LLM."""
+    sess, coord, _ui = coord_session
+    coord.task_list_remove.return_value = {
+        "error": "task_list envelope is corrupt on disk; refusing to overwrite."
+    }
+    item = sess._prepare_tool(_tc("task_list", {"action": "remove", "task_id": "x"}))
+    _, output = sess._exec_task_list(item)
+    parsed = json.loads(output)
+    assert "corrupt" in parsed["error"]
+
+
+def test_task_list_exec_remove_success_dispatches(coord_session):
+    sess, coord, _ui = coord_session
+    coord.task_list_remove.return_value = {"ok": True, "task_id": "tsk_1"}
+    item = sess._prepare_tool(_tc("task_list", {"action": "remove", "task_id": "tsk_1"}))
+    _, output = sess._exec_task_list(item)
+    parsed = json.loads(output)
+    assert parsed.get("ok") is True

tests/test_route_proxy_audit.py

@@ -0,0 +1,412 @@
+"""Tests for routing-proxy audit middleware.
+
+Every successful ``/v1/api/route/*`` hop emits an ``audit_events`` row
+with action ``route.workstream.{create,send,close,delete}`` /
+``route.{approve,cancel,command,plan}`` and ``detail`` carrying
+``{src, node_id, coord_ws_id?}``.  Failure paths (4xx/5xx) MUST NOT
+emit, and audit-emission failure MUST NOT break the proxied call.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+from unittest.mock import MagicMock
+
+import httpx
+import pytest
+from starlette.testclient import TestClient
+
+from turnstone.console.collector import ClusterCollector
+from turnstone.console.router import ConsoleRouter, NodeRef
+from turnstone.core.auth import JWT_AUD_CONSOLE, create_jwt
+
+_TEST_JWT_SECRET = "test-jwt-secret-minimum-32-chars!"
+
+
+def _coordinator_jwt(coord_ws_id: str = "coord-42") -> str:
+    """Mint a JWT shaped like CoordinatorTokenManager would produce."""
+    return create_jwt(
+        user_id="user-real-creator",
+        scopes=frozenset({"read", "write", "approve"}),
+        source="coordinator",
+        secret=_TEST_JWT_SECRET,
+        audience=JWT_AUD_CONSOLE,
+        permissions=frozenset({"admin.coordinator"}),
+        extra_claims={"coord_ws_id": coord_ws_id},
+    )
+
+
+def _plain_jwt() -> str:
+    """A normal JWT — not coordinator-origin."""
+    return create_jwt(
+        user_id="user-human",
+        scopes=frozenset({"read", "write", "approve"}),
+        source="jwt",
+        secret=_TEST_JWT_SECRET,
+        audience=JWT_AUD_CONSOLE,
+    )
+
+
+_COORD_HEADERS: dict[str, str] = {"Authorization": f"Bearer {_coordinator_jwt()}"}
+_PLAIN_HEADERS: dict[str, str] = {"Authorization": f"Bearer {_plain_jwt()}"}
+
+
+# ---------------------------------------------------------------------------
+# Mock plumbing
+# ---------------------------------------------------------------------------
+
+
+def _make_mock_collector() -> MagicMock:
+    collector = MagicMock(spec=ClusterCollector)
+    collector.get_overview.return_value = {
+        "nodes": 1,
+        "workstreams": 0,
+        "states": {"running": 0, "thinking": 0, "attention": 0, "idle": 0, "error": 0},
+        "aggregate": {"total_tokens": 0, "total_tool_calls": 0},
+    }
+    return collector
+
+
+def _make_mock_router(node_id: str = "node-a", url: str = "http://a:8080") -> MagicMock:
+    router = MagicMock(spec=ConsoleRouter)
+    router.is_ready.return_value = True
+    router.route.return_value = NodeRef(node_id, url)
+    router.generate_ws_id_for_node.return_value = "00ff" + "0" * 28
+    return router
+
+
+def _make_app(router: Any = None) -> Any:
+    from turnstone.console.server import _load_static, create_app
+
+    _load_static()
+    return create_app(
+        collector=_make_mock_collector(),
+        jwt_secret=_TEST_JWT_SECRET,
+        router=router,
+    )
+
+
+def _make_proxy(status_code: int = 200, body: dict[str, Any] | None = None) -> MagicMock:
+    payload = body or {"ws_id": "abc123", "name": "test"}
+
+    async def _post(*args: Any, **kwargs: Any) -> httpx.Response:
+        return httpx.Response(
+            status_code,
+            json=payload,
+            request=httpx.Request("POST", args[0] if args else "http://test"),
+        )
+
+    proxy = MagicMock(spec=httpx.AsyncClient)
+    proxy.post = MagicMock(side_effect=_post)
+    return proxy
+
+
+def _capture_storage() -> tuple[MagicMock, list[dict[str, Any]]]:
+    """Return a mock storage that captures record_audit_event call kwargs."""
+    captured: list[dict[str, Any]] = []
+
+    def _record(**kwargs: Any) -> None:
+        captured.append(kwargs)
+
+    storage = MagicMock()
+    storage.record_audit_event = MagicMock(side_effect=_record)
+    return storage, captured
+
+
+def _wire(app: Any, proxy: MagicMock, storage: MagicMock | None = None) -> None:
+    app.state.proxy_client = proxy
+    if storage is not None:
+        app.state.auth_storage = storage
+
+
+# ---------------------------------------------------------------------------
+# route_create
+# ---------------------------------------------------------------------------
+
+
+class TestRouteCreateAudit:
+    def test_emits_route_workstream_create_on_200_with_coordinator_origin(self):
+        router = _make_mock_router("node-a", "http://a:8080")
+        app = _make_app(router=router)
+        storage, captured = _capture_storage()
+        _wire(app, _make_proxy(200, {"ws_id": "child123", "name": "child"}), storage)
+        client = TestClient(app, raise_server_exceptions=False)
+
+        resp = client.post(
+            "/v1/api/route/workstreams/new",
+            json={"name": "child"},
+            headers=_COORD_HEADERS,
+        )
+        assert resp.status_code == 200
+        assert len(captured) == 1, captured
+        row = captured[0]
+        assert row["action"] == "route.workstream.create"
+        assert row["resource_type"] == "workstream"
+        assert row["user_id"] == "user-real-creator"
+        # body["ws_id"] is set by the handler to a fresh secrets.token_hex(16)
+        # before forwarding upstream — assert it's a 32-char hex string.
+        assert len(row["resource_id"]) == 32
+        detail = json.loads(row["detail"])
+        assert detail["src"] == "coordinator"
+        assert detail["coord_ws_id"] == "coord-42"
+        assert detail["node_id"] == "node-a"
+
+        client.close()
+
+    def test_does_not_emit_on_502(self):
+        router = _make_mock_router()
+        app = _make_app(router=router)
+        storage, captured = _capture_storage()
+        _wire(app, _make_proxy(502, {"error": "upstream"}), storage)
+        client = TestClient(app, raise_server_exceptions=False)
+
+        resp = client.post(
+            "/v1/api/route/workstreams/new",
+            json={"name": "child"},
+            headers=_COORD_HEADERS,
+        )
+        assert resp.status_code == 502
+        assert captured == []
+        client.close()
+
+    def test_does_not_emit_on_400(self):
+        router = _make_mock_router()
+        app = _make_app(router=router)
+        storage, captured = _capture_storage()
+        # No proxy needed — handler returns 400 before any upstream call.
+        proxy = MagicMock(spec=httpx.AsyncClient)
+        _wire(app, proxy, storage)
+        client = TestClient(app, raise_server_exceptions=False)
+
+        # Send invalid JSON (raw body, content-type json) — handler returns 400.
+        resp = client.post(
+            "/v1/api/route/workstreams/new",
+            content=b"not json",
+            headers={**_COORD_HEADERS, "Content-Type": "application/json"},
+        )
+        assert resp.status_code == 400
+        assert captured == []
+        client.close()
+
+    def test_503_retry_records_final_node_id(self):
+        """Audit row must reflect the node that actually served 200, not the failed first node."""
+        router = _make_mock_router()
+
+        call_count = 0
+
+        def _route(_ws_id: str) -> NodeRef:
+            nonlocal call_count
+            call_count += 1
+            if call_count <= 1:
+                return NodeRef("node-a-failed", "http://a:8080")
+            return NodeRef("node-b-retry", "http://b:8080")
+
+        router.route.side_effect = _route
+        app = _make_app(router=router)
+        storage, captured = _capture_storage()
+
+        post_count = 0
+
+        async def _post(*args: Any, **kwargs: Any) -> httpx.Response:
+            nonlocal post_count
+            post_count += 1
+            url = args[0] if args else "http://test"
+            if post_count == 1:
+                return httpx.Response(
+                    503, json={"error": "overloaded"}, request=httpx.Request("POST", url)
+                )
+            return httpx.Response(
+                200, json={"ws_id": "x", "name": "n"}, request=httpx.Request("POST", url)
+            )
+
+        proxy = MagicMock(spec=httpx.AsyncClient)
+        proxy.post = MagicMock(side_effect=_post)
+        _wire(app, proxy, storage)
+        client = TestClient(app, raise_server_exceptions=False)
+
+        resp = client.post(
+            "/v1/api/route/workstreams/new",
+            json={"name": "child"},
+            headers=_COORD_HEADERS,
+        )
+        assert resp.status_code == 200
+        assert len(captured) == 1
+        detail = json.loads(captured[0]["detail"])
+        assert detail["node_id"] == "node-b-retry"
+        client.close()
+
+    def test_no_storage_means_no_emission_no_crash(self):
+        """When auth_storage is not installed (e.g. pre-config-store tests), the new code is a no-op."""
+        router = _make_mock_router()
+        app = _make_app(router=router)
+        _wire(app, _make_proxy(200, {"ws_id": "x", "name": "n"}))  # NO storage
+        client = TestClient(app, raise_server_exceptions=False)
+
+        resp = client.post(
+            "/v1/api/route/workstreams/new",
+            json={"name": "child"},
+            headers=_COORD_HEADERS,
+        )
+        assert resp.status_code == 200  # no crash
+        client.close()
+
+
+# ---------------------------------------------------------------------------
+# route_proxy
+# ---------------------------------------------------------------------------
+
+
+class TestRouteProxyAudit:
+    @pytest.mark.parametrize(
+        "path,expected_action",
+        [
+            ("/v1/api/route/send", "route.workstream.send"),
+            ("/v1/api/route/approve", "route.approve"),
+            ("/v1/api/route/cancel", "route.cancel"),
+            ("/v1/api/route/command", "route.command"),
+            ("/v1/api/route/plan", "route.plan"),
+            ("/v1/api/route/workstreams/close", "route.workstream.close"),
+        ],
+    )
+    def test_method_to_action_mapping(self, path: str, expected_action: str):
+        router = _make_mock_router("node-x", "http://x:8080")
+        app = _make_app(router=router)
+        storage, captured = _capture_storage()
+        _wire(app, _make_proxy(200, {"status": "ok"}), storage)
+        client = TestClient(app, raise_server_exceptions=False)
+
+        resp = client.post(
+            path,
+            json={"ws_id": "abc123", "message": "hi"},
+            headers=_COORD_HEADERS,
+        )
+        assert resp.status_code == 200
+        assert len(captured) == 1
+        row = captured[0]
+        assert row["action"] == expected_action
+        assert row["resource_id"] == "abc123"
+        assert row["user_id"] == "user-real-creator"
+        detail = json.loads(row["detail"])
+        assert detail["src"] == "coordinator"
+        assert detail["coord_ws_id"] == "coord-42"
+        assert detail["node_id"] == "node-x"
+        client.close()
+
+    def test_does_not_emit_on_4xx(self):
+        # Use 403 — 404 triggers the route_proxy refresh-and-retry path
+        # which reaches into ConsoleRouter internals our MagicMock
+        # doesn't model.  403 exercises the same "non-2xx, no audit"
+        # invariant without the side effect.
+        router = _make_mock_router()
+        app = _make_app(router=router)
+        storage, captured = _capture_storage()
+        _wire(app, _make_proxy(403, {"error": "forbidden"}), storage)
+        client = TestClient(app, raise_server_exceptions=False)
+
+        resp = client.post(
+            "/v1/api/route/send",
+            json={"ws_id": "abc", "message": "hi"},
+            headers=_COORD_HEADERS,
+        )
+        assert resp.status_code == 403
+        assert captured == []
+        client.close()
+
+    def test_emits_with_plain_jwt_origin_no_coord_ws_id_in_detail(self):
+        """Non-coordinator inbound: src='jwt', no coord_ws_id key in detail."""
+        router = _make_mock_router("node-y", "http://y:8080")
+        app = _make_app(router=router)
+        storage, captured = _capture_storage()
+        _wire(app, _make_proxy(200, {"status": "ok"}), storage)
+        client = TestClient(app, raise_server_exceptions=False)
+
+        resp = client.post(
+            "/v1/api/route/send",
+            json={"ws_id": "abc", "message": "hi"},
+            headers=_PLAIN_HEADERS,
+        )
+        assert resp.status_code == 200
+        assert len(captured) == 1
+        row = captured[0]
+        assert row["action"] == "route.workstream.send"
+        assert row["user_id"] == "user-human"
+        detail = json.loads(row["detail"])
+        assert detail["src"] == "jwt"
+        assert "coord_ws_id" not in detail
+        assert detail["node_id"] == "node-y"
+        client.close()
+
+
+# ---------------------------------------------------------------------------
+# route_workstream_delete
+# ---------------------------------------------------------------------------
+
+
+class TestRouteWorkstreamDeleteAudit:
+    def test_emits_route_workstream_delete_on_200(self):
+        router = _make_mock_router("node-d", "http://d:8080")
+        app = _make_app(router=router)
+        storage, captured = _capture_storage()
+        _wire(app, _make_proxy(200, {"status": "deleted"}), storage)
+        client = TestClient(app, raise_server_exceptions=False)
+
+        resp = client.post(
+            "/v1/api/route/workstreams/delete",
+            json={"ws_id": "doomed-ws"},
+            headers=_COORD_HEADERS,
+        )
+        assert resp.status_code == 200
+        assert len(captured) == 1
+        row = captured[0]
+        assert row["action"] == "route.workstream.delete"
+        assert row["resource_id"] == "doomed-ws"
+        detail = json.loads(row["detail"])
+        assert detail["node_id"] == "node-d"
+        assert detail["src"] == "coordinator"
+        assert detail["coord_ws_id"] == "coord-42"
+        client.close()
+
+    def test_does_not_emit_on_502(self):
+        router = _make_mock_router()
+        app = _make_app(router=router)
+        storage, captured = _capture_storage()
+        _wire(app, _make_proxy(502, {"error": "down"}), storage)
+        client = TestClient(app, raise_server_exceptions=False)
+
+        resp = client.post(
+            "/v1/api/route/workstreams/delete",
+            json={"ws_id": "doomed-ws"},
+            headers=_COORD_HEADERS,
+        )
+        assert resp.status_code == 502
+        assert captured == []
+        client.close()
+
+
+# ---------------------------------------------------------------------------
+# Resilience
+# ---------------------------------------------------------------------------
+
+
+class TestAuditResilience:
+    def test_emit_swallows_storage_exception(self):
+        """If record_audit_event raises, the proxied response must still come back unchanged."""
+        router = _make_mock_router()
+        app = _make_app(router=router)
+
+        storage = MagicMock()
+        storage.record_audit_event = MagicMock(side_effect=RuntimeError("DB down"))
+
+        _wire(app, _make_proxy(200, {"status": "ok"}), storage)
+        client = TestClient(app, raise_server_exceptions=False)
+
+        resp = client.post(
+            "/v1/api/route/send",
+            json={"ws_id": "abc", "message": "hi"},
+            headers=_COORD_HEADERS,
+        )
+        # Audit failure is swallowed — proxied response still 200.
+        assert resp.status_code == 200
+        assert resp.json() == {"status": "ok"}
+        client.close()

tests/test_storage_skills_filtered.py

@@ -0,0 +1,128 @@
+"""Storage-protocol tests for ``list_skills_filtered``.
+
+Runs on both SQLite and PostgreSQL via the shared ``storage_backend``
+fixture (``conftest.py``) so the tag-substring filter and column-match
+filters are validated against both backends' ``LIKE`` semantics.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+
+def _create_skill(
+    storage: Any,
+    *,
+    template_id: str,
+    name: str,
+    category: str = "general",
+    tags: list[str] | None = None,
+    scan_status: str = "",
+    enabled: bool = True,
+    priority: int = 0,
+) -> None:
+    storage.create_prompt_template(
+        template_id=template_id,
+        name=name,
+        category=category,
+        content="",
+        variables="[]",
+        is_default=False,
+        org_id="",
+        created_by="test",
+        tags=json.dumps(tags or []),
+        priority=priority,
+        enabled=enabled,
+    )
+    if scan_status:
+        # scan_status is set by the scanner pipeline, not create_prompt_template;
+        # patch it directly so tests can fix the value.
+        with storage._conn() as conn:
+            import sqlalchemy as sa
+
+            from turnstone.core.storage._schema import prompt_templates
+
+            conn.execute(
+                sa.update(prompt_templates)
+                .where(prompt_templates.c.template_id == template_id)
+                .values(scan_status=scan_status)
+            )
+            conn.commit()
+
+
+class TestListSkillsFiltered:
+    def test_no_filters_returns_all_ordered_by_priority_then_name(self, storage):
+        _create_skill(storage, template_id="s1", name="zebra", priority=10)
+        _create_skill(storage, template_id="s2", name="alpha", priority=10)
+        _create_skill(storage, template_id="s3", name="any", priority=1)
+        rows = storage.list_skills_filtered()
+        names = [r["name"] for r in rows]
+        # priority asc (1 then 10), name asc within priority.
+        assert names == ["any", "alpha", "zebra"]
+
+    def test_category_exact_match(self, storage):
+        _create_skill(storage, template_id="s1", name="a", category="ops")
+        _create_skill(storage, template_id="s2", name="b", category="engineering")
+        _create_skill(storage, template_id="s3", name="c", category="engineering")
+        rows = storage.list_skills_filtered(category="engineering")
+        assert {r["name"] for r in rows} == {"b", "c"}
+
+    def test_tag_substring_quote_safe(self, storage):
+        # Quote-bracketed pattern: `"foo"` matches `["foo", "bar"]` but not `["foobar"]`.
+        _create_skill(storage, template_id="s1", name="m", tags=["foo", "bar"])
+        _create_skill(storage, template_id="s2", name="m2", tags=["foobar"])
+        _create_skill(storage, template_id="s3", name="m3", tags=["other"])
+        rows = storage.list_skills_filtered(tag="foo")
+        assert {r["name"] for r in rows} == {"m"}
+
+    def test_tag_filter_is_case_insensitive_on_both_backends(self, storage):
+        """SQLite LIKE is case-insensitive by default; PostgreSQL is not.
+        Normalise at the filter site so dev and prod return the same rows."""
+        _create_skill(storage, template_id="s1", name="a", tags=["GPU"])
+        _create_skill(storage, template_id="s2", name="b", tags=["cpu"])
+        assert {r["name"] for r in storage.list_skills_filtered(tag="gpu")} == {"a"}
+        assert {r["name"] for r in storage.list_skills_filtered(tag="GPU")} == {"a"}
+        assert {r["name"] for r in storage.list_skills_filtered(tag="Gpu")} == {"a"}
+        assert {r["name"] for r in storage.list_skills_filtered(tag="CPU")} == {"b"}
+
+    def test_tag_filter_escapes_like_wildcards(self, storage):
+        """Literal ``%`` / ``_`` in the tag must NOT act as SQL wildcards."""
+        _create_skill(storage, template_id="s1", name="literal", tags=["a%b"])
+        _create_skill(storage, template_id="s2", name="underscore-tag", tags=["a_b"])
+        _create_skill(storage, template_id="s3", name="decoy", tags=["axxb", "acb"])
+        # Literal `%` matches only the literal tag, not arbitrary chars.
+        assert {r["name"] for r in storage.list_skills_filtered(tag="a%b")} == {"literal"}
+        # Literal `_` matches only the literal tag, not any single char.
+        assert {r["name"] for r in storage.list_skills_filtered(tag="a_b")} == {"underscore-tag"}
+
+    def test_scan_status_filter(self, storage):
+        _create_skill(storage, template_id="s1", name="a", scan_status="clean")
+        _create_skill(storage, template_id="s2", name="b", scan_status="flagged")
+        _create_skill(storage, template_id="s3", name="c")
+        rows = storage.list_skills_filtered(scan_status="flagged")
+        assert {r["name"] for r in rows} == {"b"}
+
+    def test_enabled_only_filter(self, storage):
+        _create_skill(storage, template_id="s1", name="a", enabled=True)
+        _create_skill(storage, template_id="s2", name="b", enabled=False)
+        rows = storage.list_skills_filtered(enabled_only=True)
+        assert {r["name"] for r in rows} == {"a"}
+
+    def test_limit_caps_rows(self, storage):
+        for i in range(5):
+            _create_skill(storage, template_id=f"s{i}", name=f"sk-{i:02d}")
+        rows = storage.list_skills_filtered(limit=2)
+        assert len(rows) == 2
+
+    def test_filters_combine_with_and_semantics(self, storage):
+        _create_skill(storage, template_id="s1", name="a", category="ops", tags=["alpha"])
+        _create_skill(storage, template_id="s2", name="b", category="ops", tags=["beta"])
+        _create_skill(storage, template_id="s3", name="c", category="other", tags=["alpha"])
+        rows = storage.list_skills_filtered(category="ops", tag="alpha")
+        assert {r["name"] for r in rows} == {"a"}
+
+    def test_empty_result_for_no_match(self, storage):
+        _create_skill(storage, template_id="s1", name="a", category="ops")
+        rows = storage.list_skills_filtered(category="nonexistent")
+        assert rows == []

tests/test_tools_schema.py

@@ -72,8 +72,8 @@ class TestToolsMetadata:
     """Validate the metadata extracted from JSON files."""
 
     def test_tool_count(self):
-        # 19 original tools + 6 coordinator tools
-        assert len(TOOLS) == 25
+        # 19 interactive tools + 9 coordinator tools
+        assert len(TOOLS) == 28
 
     def test_agent_tools_count(self):
         assert len(AGENT_TOOLS) == 10
@@ -84,7 +84,7 @@ class TestToolsMetadata:
     def test_coordinator_tools_count(self):
         from turnstone.core.tools import COORDINATOR_TOOLS
 
-        assert len(COORDINATOR_TOOLS) == 6
+        assert len(COORDINATOR_TOOLS) == 9
         assert {t["function"]["name"] for t in COORDINATOR_TOOLS} == {
             "spawn_workstream",
             "inspect_workstream",
@@ -92,6 +92,9 @@ class TestToolsMetadata:
             "close_workstream",
             "delete_workstream",
             "list_workstreams",
+            "list_nodes",
+            "list_skills",
+            "task_list",
         }
 
     def test_auto_approve_sets_match(self):
@@ -107,6 +110,8 @@ class TestToolsMetadata:
             # Coordinator read-only tools (no-mutation, safe to auto-approve):
             "inspect_workstream",
             "list_workstreams",
+            "list_nodes",
+            "list_skills",
         }
         assert expected == AGENT_AUTO_TOOLS
         assert expected == TASK_AUTO_TOOLS
@@ -138,6 +143,7 @@ class TestToolsMetadata:
             "send_to_workstream": "message",
             "close_workstream": "ws_id",
             "delete_workstream": "ws_id",
+            "task_list": "action",
         }
         assert expected == PRIMARY_KEY_MAP
 

tests/test_workstream_kind.py

@@ -257,6 +257,9 @@ def test_chatsession_interactive_kind_excludes_coordinator_tools(tmp_db):
         "close_workstream",
         "delete_workstream",
         "list_workstreams",
+        "list_nodes",
+        "list_skills",
+        "task_list",
     ):
         assert coord_name not in names, f"{coord_name} leaked into interactive session tools"
 

turnstone/api/console_schemas.py

@@ -944,3 +944,14 @@ class SetNodeMetadataRequest(BaseModel):
 
 class BulkSetNodeMetadataRequest(BaseModel):
     entries: list[SetNodeMetadataRequest] = Field(default_factory=list)
+
+
+class CoordinatorOpenResponse(BaseModel):
+    """Response body for POST /v1/api/coordinator/{ws_id}/open."""
+
+    ws_id: str
+    name: str
+    already_loaded: bool | None = Field(
+        default=None,
+        description="True when the coordinator was already in memory; absent after a fresh rehydrate.",
+    )

turnstone/api/console_spec.py

@@ -21,6 +21,7 @@ from turnstone.api.console_schemas import (
     ConsoleCreateWsRequest,
     ConsoleCreateWsResponse,
     ConsoleHealthResponse,
+    CoordinatorOpenResponse,
     CreateChannelUserRequest,
     CreateMcpServerRequest,
     CreateModelDefinitionRequest,
@@ -1106,6 +1107,15 @@ CONSOLE_ENDPOINTS: list[EndpointSpec] = [
         error_codes=[400, 503],
         tags=["Routing"],
     ),
+    # --- Coordinator workstream API ---
+    EndpointSpec(
+        "/v1/api/coordinator/{ws_id}/open",
+        "POST",
+        "Open (rehydrate) a coordinator workstream by ws_id",
+        response_model=CoordinatorOpenResponse,
+        error_codes=[400, 403, 404, 500, 503],
+        tags=["Coordinator"],
+    ),
     # --- Observability ---
     EndpointSpec(
         "/health",
@@ -1142,6 +1152,7 @@ _ALL_MODELS: list[type[BaseModel]] = [
     ConsoleCreateWsRequest,
     ConsoleCreateWsResponse,
     ConsoleHealthResponse,
+    CoordinatorOpenResponse,
     CreateScheduleRequest,
     UpdateScheduleRequest,
     ScheduleInfo,

turnstone/console/coordinator_client.py

@@ -23,8 +23,11 @@ JWTs carrying the real user's identity + scopes.
 
 from __future__ import annotations
 
+import json
+import secrets
 import threading
 import time
+from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any
 
 import httpx
@@ -32,6 +35,18 @@ import httpx
 from turnstone.core.auth import JWT_AUD_CONSOLE, create_jwt
 from turnstone.core.log import get_logger
 
+_TASK_STATUSES = frozenset({"pending", "in_progress", "done", "blocked"})
+# Hard cap on tasks per coordinator — the full list is read and re-serialized
+# on every mutation, so unbounded growth is both a storage and a tool-output-size
+# hazard.  Hitting the cap is an explicit signal to prune done/blocked rows.
+_TASK_LIST_MAX = 500
+
+
+def _utc_now_iso() -> str:
+    """ISO-8601 UTC timestamp with seconds precision — used for task timestamps."""
+    return datetime.now(UTC).replace(microsecond=0).isoformat()
+
+
 if TYPE_CHECKING:
     from collections.abc import Callable
 
@@ -154,6 +169,12 @@ class CoordinatorClient:
         # with the coordinator session.
         self._http = http_client or httpx.Client(timeout=timeout)
         self._owns_http = http_client is None
+        # task_list per-ws lock cache — populated lazily by _task_lock().
+        # Single-session so a plain dict behind a coarse lock is fine;
+        # WeakValueDictionary isn't needed (entries live as long as the
+        # CoordinatorClient instance).
+        self._task_lock_cache: dict[str, threading.Lock] = {}
+        self._task_lock_cache_lock = threading.Lock()
 
     # -- lifecycle ----------------------------------------------------------
 
@@ -338,6 +359,304 @@ class CoordinatorClient:
         truncated = len(raw) >= limit
         return {"children": children, "truncated": truncated}
 
+    def list_nodes(
+        self,
+        *,
+        filters: dict[str, Any] | None = None,
+        limit: int = 100,
+    ) -> dict[str, Any]:
+        """Return ``{"nodes": [...], "truncated": bool}``.
+
+        Each row carries the node's full metadata dict — both auto-populated
+        keys (``arch``, ``cpu_count``, ``fqdn``, ``hostname``, ``os``,
+        ``os_release``, ``python``; always present, ``source="auto"``) and
+        operator-supplied user keys (deployment-specific, ``source="user"``).
+        ``filters`` matches all key=value pairs (AND semantics) and is
+        pushed into SQL via ``filter_nodes_by_metadata`` — no per-row
+        lookups.
+
+        Storage stores metadata values as JSON-encoded strings (the write
+        path in ``server.py`` / ``admin.py`` / ``console/server.py`` all
+        go through ``json.dumps``).  Filter values get re-encoded so the
+        stored-text comparison succeeds, and read values get decoded so
+        the model sees the natural Python form (``"x86_64"`` not
+        ``'"x86_64"'``, ``4`` not ``"4"``).
+        """
+        page_size = max(1, min(int(limit), 500))
+        if filters:
+            # Filtered case: narrow to the matching ids first, then pull
+            # metadata only for the ``page_size``-bounded slice.  Avoids
+            # the full-cluster ``get_all_node_metadata`` scan when the
+            # model is asking for a handful of nodes.  Per-node lookups
+            # are bounded at 500 by the limit clamp.
+            encoded_filters = {str(k): json.dumps(v) for k, v in filters.items()}
+            matching = self._storage.filter_nodes_by_metadata(encoded_filters)
+            node_ids = sorted(matching)
+            truncated = len(node_ids) > page_size
+            node_ids = node_ids[:page_size]
+            meta_rows_by_node: dict[str, list[dict[str, Any]]] = {
+                nid: self._storage.get_node_metadata(nid) for nid in node_ids
+            }
+        else:
+            # Unfiltered case: one wide query.  The caller is paging
+            # through the whole cluster and needs metadata for every
+            # node anyway — per-node lookups would be a true N+1.
+            all_meta = self._storage.get_all_node_metadata()
+            node_ids = sorted(all_meta.keys())
+            truncated = len(node_ids) > page_size
+            node_ids = node_ids[:page_size]
+            meta_rows_by_node = {nid: all_meta.get(nid, []) for nid in node_ids}
+        nodes: list[dict[str, Any]] = []
+        for nid in node_ids:
+            meta: dict[str, dict[str, Any]] = {}
+            for r in meta_rows_by_node.get(nid, []):
+                key = r.get("key")
+                if not key:
+                    continue
+                raw_value = r.get("value", "")
+                try:
+                    decoded = json.loads(raw_value) if isinstance(raw_value, str) else raw_value
+                except (TypeError, ValueError):
+                    decoded = raw_value
+                meta[str(key)] = {
+                    "value": decoded,
+                    "source": str(r.get("source", "")),
+                }
+            nodes.append({"node_id": nid, "metadata": meta})
+        return {"nodes": nodes, "truncated": truncated}
+
+    def list_skills(
+        self,
+        *,
+        category: str | None = None,
+        tag: str | None = None,
+        scan_status: str | None = None,
+        enabled_only: bool = False,
+        limit: int = 100,
+    ) -> dict[str, Any]:
+        """Return ``{"skills": [...], "truncated": bool}``.
+
+        Filters pushed into SQL via ``list_skills_filtered`` — no per-row
+        lookups.  ``tag`` matches when the value appears in the
+        JSON-array ``tags`` column (quote-bracketed substring).
+        ``tags`` is decoded from JSON at the edge so the model sees a
+        list, not the escaped string.  Projection is intentionally narrow
+        — discovery metadata only, not full row.
+        """
+        page_size = max(1, min(int(limit), 500))
+        rows = self._storage.list_skills_filtered(
+            category=category,
+            tag=tag,
+            scan_status=scan_status,
+            enabled_only=enabled_only,
+            limit=page_size + 1,  # +1 to detect truncation
+        )
+        truncated = len(rows) > page_size
+        rows = rows[:page_size]
+        skills: list[dict[str, Any]] = []
+        for r in rows:
+            tags_raw = r.get("tags") or "[]"
+            try:
+                tags = json.loads(tags_raw) if isinstance(tags_raw, str) else list(tags_raw)
+            except (TypeError, ValueError):
+                tags = []
+            skills.append(
+                {
+                    "name": r.get("name") or "",
+                    "category": r.get("category") or "",
+                    "tags": tags,
+                    "version": r.get("version") or "",
+                    "description": r.get("description") or "",
+                    "model": r.get("model") or "",
+                    "enabled": bool(r.get("enabled")),
+                    "scan_status": r.get("scan_status") or "",
+                    "activation": r.get("activation") or "",
+                }
+            )
+        return {"skills": skills, "truncated": truncated}
+
+    # ------------------------------------------------------------------
+    # task_list — coordinator-local planning state persisted on workstream_config
+    # ------------------------------------------------------------------
+
+    def task_list_get(self, ws_id: str) -> dict[str, Any]:
+        """Return the task envelope ``{"version": 1, "tasks": [...]}``.
+
+        Corrupt / legacy config rows return an empty envelope rather than
+        raising — a hand-edited DB shouldn't break the read path.  The
+        mutating methods use ``_load_task_envelope_strict`` to detect
+        corruption and refuse to overwrite silently.
+        """
+        env, _ = self._load_task_envelope(ws_id)
+        return env
+
+    def _load_task_envelope(self, ws_id: str) -> tuple[dict[str, Any], bool]:
+        """Return ``(envelope, corrupt)``; ``corrupt=True`` iff the stored
+        payload is non-empty and unparseable as the expected shape."""
+        empty: dict[str, Any] = {"version": 1, "tasks": []}
+        if ws_id != self._coord_ws_id:
+            return empty, False
+        raw = self._storage.load_workstream_config(ws_id) or {}
+        payload = raw.get("tasks")
+        if not payload:
+            return empty, False
+        try:
+            data = json.loads(payload)
+        except (TypeError, ValueError):
+            log.warning("task_list.corrupt_envelope ws=%s (unparseable JSON)", ws_id)
+            return empty, True
+        if not (isinstance(data, dict) and isinstance(data.get("tasks"), list)):
+            log.warning("task_list.corrupt_envelope ws=%s (wrong shape)", ws_id)
+            return empty, True
+        return data, False
+
+    def _save_task_list(self, ws_id: str, envelope: dict[str, Any]) -> None:
+        # Save only the ``tasks`` key so concurrent writers to other
+        # workstream_config keys (e.g. reasoning_effort from the admin UI)
+        # aren't clobbered by a read-modify-write on the full row.
+        self._storage.save_workstream_config(
+            ws_id, {"tasks": json.dumps(envelope, separators=(",", ":"))}
+        )
+
+    def _task_lock(self, ws_id: str) -> threading.Lock:
+        """Per-ws lock cached on the client.
+
+        Coordinator tool execs run on a single worker thread so contention
+        is unlikely in practice, but the lock is cheap defence-in-depth
+        for any future caller (maintenance script, HTTP handler) that
+        mutates the list outside the worker thread.
+        """
+        with self._task_lock_cache_lock:
+            lk = self._task_lock_cache.get(ws_id)
+            if lk is None:
+                lk = threading.Lock()
+                self._task_lock_cache[ws_id] = lk
+            return lk
+
+    def task_list_add(
+        self,
+        ws_id: str,
+        *,
+        title: str,
+        status: str = "pending",
+        child_ws_id: str = "",
+    ) -> dict[str, Any]:
+        if ws_id != self._coord_ws_id:
+            return {"error": f"task_list scope violation: {ws_id}"}
+        clean_title = (title or "").strip()[:200]
+        if not clean_title:
+            return {"error": "title is required"}
+        if status not in _TASK_STATUSES:
+            return {"error": f"invalid status: {status}"}
+        with self._task_lock(ws_id):
+            envelope, corrupt = self._load_task_envelope(ws_id)
+            if corrupt:
+                return {
+                    "error": (
+                        "task_list envelope is corrupt on disk; refusing to "
+                        "overwrite.  Inspect workstream_config.tasks manually "
+                        "or clear it before retrying."
+                    )
+                }
+            if len(envelope["tasks"]) >= _TASK_LIST_MAX:
+                return {
+                    "error": (
+                        f"task_list capacity reached ({_TASK_LIST_MAX}).  "
+                        "Remove completed tasks before adding more."
+                    )
+                }
+            now = _utc_now_iso()
+            task = {
+                "id": "tsk_" + secrets.token_hex(6),
+                "title": clean_title,
+                "status": status,
+                "child_ws_id": child_ws_id,
+                "created": now,
+                "updated": now,
+            }
+            envelope["tasks"].append(task)
+            self._save_task_list(ws_id, envelope)
+            return task
+
+    def task_list_update(
+        self,
+        ws_id: str,
+        *,
+        task_id: str,
+        title: str | None = None,
+        status: str | None = None,
+        child_ws_id: str | None = None,
+    ) -> dict[str, Any]:
+        if ws_id != self._coord_ws_id:
+            return {"error": f"task_list scope violation: {ws_id}"}
+        if status is not None and status not in _TASK_STATUSES:
+            return {"error": f"invalid status: {status}"}
+        with self._task_lock(ws_id):
+            envelope, corrupt = self._load_task_envelope(ws_id)
+            if corrupt:
+                return {"error": ("task_list envelope is corrupt on disk; refusing to overwrite.")}
+            for t in envelope["tasks"]:
+                if t.get("id") == task_id:
+                    if title is not None:
+                        clean = title.strip()[:200]
+                        if not clean:
+                            return {"error": "title cannot be empty"}
+                        t["title"] = clean
+                    if status is not None:
+                        t["status"] = status
+                    if child_ws_id is not None:
+                        t["child_ws_id"] = child_ws_id
+                    t["updated"] = _utc_now_iso()
+                    self._save_task_list(ws_id, envelope)
+                    # t is a dict pulled out of a json-decoded list; mypy
+                    # sees it as Any from the decode path.  Cast back to
+                    # the annotated return type.
+                    return dict(t)
+            return {"error": f"task not found: {task_id}"}
+
+    def task_list_remove(self, ws_id: str, *, task_id: str) -> dict[str, Any]:
+        """Remove a task by id.  Returns a result dict shaped like the
+        other mutators — the caller can then distinguish scope violation
+        vs corrupt envelope vs genuine not-found rather than collapsing
+        all three into ``False`` (which would mis-report a corrupt DB
+        as "task not found" to the coordinator LLM).
+        """
+        if ws_id != self._coord_ws_id:
+            return {"error": f"task_list scope violation: {ws_id}"}
+        with self._task_lock(ws_id):
+            envelope, corrupt = self._load_task_envelope(ws_id)
+            if corrupt:
+                return {"error": ("task_list envelope is corrupt on disk; refusing to overwrite.")}
+            before = len(envelope["tasks"])
+            envelope["tasks"] = [t for t in envelope["tasks"] if t.get("id") != task_id]
+            if len(envelope["tasks"]) == before:
+                return {"error": f"task not found: {task_id}"}
+            self._save_task_list(ws_id, envelope)
+            return {"ok": True, "task_id": task_id}
+
+    def task_list_reorder(self, ws_id: str, *, task_ids: list[str]) -> dict[str, Any]:
+        """Reject unless ``task_ids`` is an exact permutation of the
+        current set — prevents silent task loss from a partial reorder.
+        """
+        if ws_id != self._coord_ws_id:
+            return {"error": f"task_list scope violation: {ws_id}"}
+        with self._task_lock(ws_id):
+            envelope, corrupt = self._load_task_envelope(ws_id)
+            if corrupt:
+                return {"error": ("task_list envelope is corrupt on disk; refusing to overwrite.")}
+            current = [t.get("id") for t in envelope["tasks"]]
+            if set(task_ids) != set(current) or len(task_ids) != len(current):
+                return {
+                    "error": (
+                        "task_ids must be a permutation of the existing set. "
+                        f"current={sorted(filter(None, current))}"
+                    ),
+                }
+            by_id = {t.get("id"): t for t in envelope["tasks"]}
+            envelope["tasks"] = [by_id[tid] for tid in task_ids]
+            self._save_task_list(ws_id, envelope)
+            return {"ok": True, "order": task_ids}
+
     def inspect(self, ws_id: str, *, message_limit: int = 20) -> dict[str, Any]:
         """Return persisted workstream state + tail-N messages + recent verdicts.
 

turnstone/console/server.py

@@ -216,6 +216,65 @@ def _proxy_auth_headers(request: Request) -> dict[str, str]:
     return {}
 
 
+# Action-name map for the routing proxy.  See ``turnstone/core/audit.py``
+# module docstring for the canonical action-namespace registry.
+_ROUTE_PROXY_AUDIT_ACTIONS: dict[str, str] = {
+    "send": "route.workstream.send",
+    "approve": "route.approve",
+    "cancel": "route.cancel",
+    "command": "route.command",
+    "plan": "route.plan",
+    "close": "route.workstream.close",
+}
+
+
+def _emit_route_audit(
+    request: Request,
+    action: str,
+    ws_id: str,
+    node_id: str,
+) -> None:
+    """Record an audit event for a successful routing-proxy hop.
+
+    Caller must ensure the upstream response was 2xx; auditing failures
+    is deferred (4xx/5xx are observable via ``_record_route``'s metrics
+    path).  Reads the inbound ``auth_result`` directly so that
+    coordinator-origin attribution lands in ``detail.src`` without
+    relying on the ``_proxy_auth_headers`` re-mint.
+
+    ``detail`` carries ``{src, node_id, coord_ws_id?}`` — ``coord_ws_id``
+    only when the inbound JWT carried it (i.e. the call originated from
+    a coordinator session).  Failures are swallowed; the proxied
+    response must never break because of an audit-emission bug.
+    """
+    storage = getattr(request.app.state, "auth_storage", None)
+    if storage is None:
+        return
+    auth = getattr(getattr(request, "state", None), "auth_result", None)
+    user_id: str = (getattr(auth, "user_id", "") or "") if auth is not None else ""
+    src: str = (getattr(auth, "token_source", "") or "") if auth is not None else ""
+    coord_ws_id: str = ""
+    if auth is not None:
+        coord_ws_id = (getattr(auth, "extra_claims", None) or {}).get("coord_ws_id", "") or ""
+    detail: dict[str, Any] = {"src": src, "node_id": node_id}
+    if coord_ws_id:
+        detail["coord_ws_id"] = coord_ws_id
+    try:
+        from turnstone.core.audit import record_audit
+
+        record_audit(
+            storage,
+            user_id,
+            action,
+            "workstream",
+            ws_id,
+            detail,
+            request.client.host if request.client else "",
+        )
+    except Exception:
+        log.debug("route.audit_failed action=%s", action, exc_info=True)
+
+
 def _get_server_url(request: Request, node_id: str) -> str | None:
     """Resolve node_id to its server_url via the collector."""
     if not node_id or not _VALID_NODE_ID.match(node_id) or len(node_id) > 256:
@@ -833,6 +892,15 @@ async def route_create(request: Request) -> Response:
         data = resp.json()
         data["node_url"] = ref.url
         data["node_id"] = ref.node_id
+        # Audit attribution — multipart sets ``ws_id`` from the query
+        # string; JSON sets it on the body (or carries ``resume_ws``
+        # for a rehydrate).  Either way, this is the workstream the
+        # caller actually landed on.
+        if is_multipart:
+            audit_ws_id = ws_id
+        else:
+            audit_ws_id = body.get("ws_id") or body.get("resume_ws", "") or ""
+        _emit_route_audit(request, "route.workstream.create", audit_ws_id, ref.node_id)
         return _record_route(request, "create", 200, t0, JSONResponse(data))
     return _record_route(
         request,
@@ -1094,7 +1162,12 @@ async def route_proxy(request: Request) -> Response:
                         status_code=502,
                     ),
                 )
+            ref = new_ref  # retried node — used for audit attribution (only emits on 2xx via the next block).
 
+    if 200 <= resp.status_code < 300:
+        action = _ROUTE_PROXY_AUDIT_ACTIONS.get(method)
+        if action:
+            _emit_route_audit(request, action, ws_id, ref.node_id)
     return _record_route(
         request,
         method,
@@ -1185,6 +1258,8 @@ async def route_workstream_delete(request: Request) -> Response:
             ),
         )
 
+    if 200 <= resp.status_code < 300:
+        _emit_route_audit(request, "route.workstream.delete", ws_id, ref.node_id)
     return _record_route(
         request,
         "delete",
@@ -1982,6 +2057,57 @@ async def coordinator_detail(request: Request) -> JSONResponse:
     )
 
 
+async def coordinator_open(request: Request) -> JSONResponse:
+    """POST /v1/api/coordinator/{ws_id}/open — explicit rehydration.
+
+    Parity with the server's ``POST /v1/api/workstreams/{ws_id}/open``.
+    ``coordinator_detail`` already rehydrates lazily on a GET miss; this
+    endpoint gives SDK callers and operators a way to warm a coordinator
+    without browsing to it.  Same ownership / 404-on-mismatch /
+    correlation-id-masked error semantics as ``coordinator_detail``.
+    """
+    err = _require_admin_coordinator(request)
+    if err is not None:
+        return err
+    coord_mgr, err503 = _require_coord_mgr(request)
+    if err503 is not None:
+        return err503
+    ws_id = request.path_params.get("ws_id", "")
+    if not ws_id:
+        return JSONResponse({"error": "ws_id is required"}, status_code=400)
+    user_id = _auth_user_id(request)
+    ws = coord_mgr.get(ws_id)
+    if ws is not None:
+        if ws.user_id != user_id and not _is_admin(request):
+            return JSONResponse({"error": "coordinator not found"}, status_code=404)
+        return JSONResponse({"ws_id": ws.id, "name": ws.name, "already_loaded": True})
+    try:
+        ws = coord_mgr.open_admin(ws_id) if _is_admin(request) else coord_mgr.open(ws_id, user_id)
+    except ValueError as exc:
+        return JSONResponse({"error": str(exc)}, status_code=503)
+    except Exception:
+        correlation_id = secrets.token_hex(4)
+        log.warning(
+            "coordinator_open.rehydrate_failed correlation_id=%s ws_id=%s",
+            correlation_id,
+            ws_id[:8],
+            exc_info=True,
+        )
+        return JSONResponse(
+            {
+                "error": (
+                    f"failed to open coordinator (internal error). correlation_id={correlation_id}"
+                )
+            },
+            status_code=500,
+        )
+    if ws is None:
+        return JSONResponse({"error": "coordinator not found"}, status_code=404)
+    if ws.user_id != user_id and not _is_admin(request):
+        return JSONResponse({"error": "coordinator not found"}, status_code=404)
+    return JSONResponse({"ws_id": ws.id, "name": ws.name})
+
+
 # ---------------------------------------------------------------------------
 # Lifespan
 # ---------------------------------------------------------------------------
@@ -8384,6 +8510,11 @@ def create_app(
                         coordinator_close,
                         methods=["POST"],
                     ),
+                    Route(
+                        "/api/coordinator/{ws_id}/open",
+                        coordinator_open,
+                        methods=["POST"],
+                    ),
                     Route(
                         "/api/coordinator/{ws_id}/events",
                         coordinator_events,

turnstone/core/audit.py

@@ -2,6 +2,37 @@
 
 Provides a fire-and-forget ``record_audit`` function that admin handlers
 call after mutations to create a persistent audit trail.
+
+Action-name conventions (non-exhaustive — grep
+``record_audit(`` in the tree for the live set):
+
+    coordinator.*       console-side coordinator lifecycle
+                        (``coordinator.create`` / ``.close`` /
+                        ``.cancel``).
+
+    route.*             multi-node routing proxy hops
+                        (``route.workstream.create`` / ``.send`` /
+                        ``.close`` / ``.delete``,
+                        ``route.approve`` / ``.cancel`` /
+                        ``.command`` / ``.plan``).  ``detail`` carries
+                        ``{src, node_id, coord_ws_id?}`` so coordinator-
+                        origin attribution is preserved without a
+                        schema migration.
+
+    <resource>.<verb>   per-resource CRUD on admin handlers — verbs
+                        are typically ``create`` / ``update`` /
+                        ``delete``.  Resource prefixes in tree today
+                        include ``user``, ``role``, ``policy``,
+                        ``skill``, ``skill_resource``,
+                        ``oidc_identity``, ``mcp_server``,
+                        ``model_definition``, ``channel``,
+                        ``heuristic_rule``, ``output_guard_pattern``,
+                        ``prompt_policy``, ``setting``, ``token``,
+                        ``conversation``, ``memory``, ``org``.
+
+When adding a new namespace, prefer extending an existing prefix over
+inventing a synonym (e.g. ``mcp_server.refresh`` rather than
+``mcp.refresh`` — ``mcp_server.*`` is already the established prefix).
 """
 
 from __future__ import annotations

turnstone/core/session.py

@@ -3794,6 +3794,7 @@ class ChatSession:
             # included so malformed coordinator tool calls aren't a
             # dead-end.
             for key in (
+                "action",
                 "command",
                 "code",
                 "content",
@@ -3806,6 +3807,9 @@ class ChatSession:
                 "pattern",
                 "prompt",
                 "query",
+                "status",
+                "task_id",
+                "title",
                 "uri",
                 "url",
                 "ws_id",
@@ -3882,6 +3886,9 @@ class ChatSession:
             "close_workstream": self._prepare_close_workstream,
             "delete_workstream": self._prepare_delete_workstream,
             "list_workstreams": self._prepare_list_workstreams,
+            "list_nodes": self._prepare_list_nodes,
+            "list_skills": self._prepare_list_skills,
+            "task_list": self._prepare_task_list,
         }
         preparer = preparers.get(func_name)
         if not preparer:
@@ -4813,6 +4820,41 @@ class ChatSession:
             "error": f"Error: {msg}",
         }
 
+    @staticmethod
+    def _coord_str_arg(args: dict[str, Any], key: str, default: str = "") -> str:
+        """Return ``args[key]`` if it's a string, else ``default``.
+
+        Coordinator tool args come from an LLM and may be ill-typed
+        (int / list / dict in a string slot).  A naive
+        ``(args.get(key) or "").strip()`` raises ``AttributeError`` on
+        such inputs and kills the whole tool call; this guard lets the
+        prepare layer fall through to its own "required" validation and
+        produce a clean error item instead.
+        """
+        val = args.get(key)
+        return val if isinstance(val, str) else default
+
+    @staticmethod
+    def _coord_bool_arg(args: dict[str, Any], key: str, default: bool = False) -> bool:
+        """Return ``args[key]`` as a bool with robust string coercion.
+
+        Plain ``bool(x)`` treats ``"false"`` as truthy (non-empty string).
+        Accept actual bools verbatim; parse common string forms; return
+        ``default`` for anything else.
+        """
+        val = args.get(key)
+        if isinstance(val, bool):
+            return val
+        if isinstance(val, str):
+            normalized = val.strip().lower()
+            if normalized in ("true", "1", "yes", "on"):
+                return True
+            if normalized in ("false", "0", "no", "off", ""):
+                return False
+        if isinstance(val, (int, float)) and not isinstance(val, bool):
+            return bool(val)
+        return default
+
     def _prepare_spawn_workstream(self, call_id: str, args: dict[str, Any]) -> dict[str, Any]:
         if self._coord_client is None:
             return self._coord_tool_error(
@@ -5136,6 +5178,295 @@ class ChatSession:
         self._report_tool_result(call_id, "list_workstreams", summary)
         return call_id, self._truncate_output(output)
 
+    def _prepare_list_nodes(self, call_id: str, args: dict[str, Any]) -> dict[str, Any]:
+        if self._coord_client is None:
+            return self._coord_tool_error(call_id, "list_nodes", "coordinator client unavailable")
+        raw_filters = args.get("filters")
+        # Metadata values are JSON-encoded at rest; the client handles the
+        # encode/decode so preserve the model's natural types (``4`` stays
+        # an int, ``"gpu"`` stays a string) rather than stringifying here.
+        filters: dict[str, Any] = {}
+        if isinstance(raw_filters, dict):
+            for k, v in raw_filters.items():
+                if isinstance(k, str) and k and isinstance(v, (str, int, float, bool)):
+                    filters[k] = v
+        try:
+            limit = int(args.get("limit") or 100)
+        except (TypeError, ValueError):
+            limit = 100
+        limit = max(1, min(limit, 500))
+        header_bits = ["\u2699 list_nodes"]
+        if filters:
+            header_bits.append(
+                "filters=" + ",".join(f"{k}={v}" for k, v in sorted(filters.items()))
+            )
+        return {
+            "call_id": call_id,
+            "func_name": "list_nodes",
+            "header": " ".join(header_bits),
+            "preview": "",
+            "needs_approval": False,
+            "execute": self._exec_list_nodes,
+            "filters": filters,
+            "limit": limit,
+        }
+
+    def _exec_list_nodes(self, item: dict[str, Any]) -> tuple[str, str]:
+        call_id = item["call_id"]
+        try:
+            result = self._coord_client.list_nodes(
+                filters=item["filters"] or None,
+                limit=item["limit"],
+            )
+        except Exception as e:
+            msg = f"Error: list_nodes failed: {e}"
+            self._report_tool_result(call_id, "list_nodes", msg, is_error=True)
+            return call_id, msg
+        nodes = result.get("nodes", [])
+        truncated = bool(result.get("truncated"))
+        output = json.dumps(
+            {"nodes": nodes, "truncated": truncated},
+            separators=(",", ":"),
+            default=str,
+        )
+        summary = f"{len(nodes)} nodes"
+        if truncated:
+            summary += " (truncated — narrow filters or raise limit)"
+        self._report_tool_result(call_id, "list_nodes", summary)
+        return call_id, self._truncate_output(output)
+
+    def _prepare_list_skills(self, call_id: str, args: dict[str, Any]) -> dict[str, Any]:
+        if self._coord_client is None:
+            return self._coord_tool_error(call_id, "list_skills", "coordinator client unavailable")
+        category = self._coord_str_arg(args, "category").strip() or None
+        tag = self._coord_str_arg(args, "tag").strip() or None
+        scan_status = self._coord_str_arg(args, "scan_status").strip() or None
+        enabled_only = self._coord_bool_arg(args, "enabled_only")
+        try:
+            limit = int(args.get("limit") or 100)
+        except (TypeError, ValueError):
+            limit = 100
+        limit = max(1, min(limit, 500))
+        header_bits = ["\u2699 list_skills"]
+        if category:
+            header_bits.append(f"category={category}")
+        if tag:
+            header_bits.append(f"tag={tag}")
+        if scan_status:
+            header_bits.append(f"scan_status={scan_status}")
+        if enabled_only:
+            header_bits.append("enabled_only=true")
+        return {
+            "call_id": call_id,
+            "func_name": "list_skills",
+            "header": " ".join(header_bits),
+            "preview": "",
+            "needs_approval": False,
+            "execute": self._exec_list_skills,
+            "category": category,
+            "tag": tag,
+            "scan_status": scan_status,
+            "enabled_only": enabled_only,
+            "limit": limit,
+        }
+
+    def _exec_list_skills(self, item: dict[str, Any]) -> tuple[str, str]:
+        call_id = item["call_id"]
+        try:
+            result = self._coord_client.list_skills(
+                category=item["category"],
+                tag=item["tag"],
+                scan_status=item["scan_status"],
+                enabled_only=item["enabled_only"],
+                limit=item["limit"],
+            )
+        except Exception as e:
+            msg = f"Error: list_skills failed: {e}"
+            self._report_tool_result(call_id, "list_skills", msg, is_error=True)
+            return call_id, msg
+        skills = result.get("skills", [])
+        truncated = bool(result.get("truncated"))
+        output = json.dumps(
+            {"skills": skills, "truncated": truncated},
+            separators=(",", ":"),
+            default=str,
+        )
+        summary = f"{len(skills)} skills"
+        if truncated:
+            summary += " (truncated — narrow filters or raise limit)"
+        self._report_tool_result(call_id, "list_skills", summary)
+        return call_id, self._truncate_output(output)
+
+    def _prepare_task_list(self, call_id: str, args: dict[str, Any]) -> dict[str, Any]:
+        """Prepare a task_list action — list is auto-approved, mutations gated."""
+        if self._coord_client is None:
+            return self._coord_tool_error(call_id, "task_list", "coordinator client unavailable")
+        action = self._coord_str_arg(args, "action").strip().lower()
+        if action not in {"add", "update", "remove", "reorder", "list"}:
+            return self._coord_tool_error(
+                call_id,
+                "task_list",
+                "action must be one of: add, update, remove, reorder, list",
+            )
+        if action == "list":
+            return {
+                "call_id": call_id,
+                "func_name": "task_list",
+                "header": "\u2699 task_list list",
+                "preview": "",
+                "needs_approval": False,
+                "execute": self._exec_task_list,
+                "action": "list",
+            }
+        # --- mutating actions -------------------------------------------------
+        item: dict[str, Any] = {
+            "call_id": call_id,
+            "func_name": "task_list",
+            "needs_approval": True,
+            "execute": self._exec_task_list,
+            "action": action,
+        }
+        if action == "add":
+            # Reject non-string title / status / child_ws_id up front so
+            # a malformed model call (``title=42``) produces a clean tool
+            # error rather than an AttributeError during ``.strip()``.
+            for field_name in ("title", "status", "child_ws_id"):
+                raw = args.get(field_name)
+                if raw is not None and not isinstance(raw, str):
+                    return self._coord_tool_error(
+                        call_id, "task_list", f"add: {field_name} must be a string"
+                    )
+            title = self._coord_str_arg(args, "title").strip()
+            if not title:
+                return self._coord_tool_error(call_id, "task_list", "add: title is required")
+            status = self._coord_str_arg(args, "status", "pending").strip() or "pending"
+            child_ws_id = self._coord_str_arg(args, "child_ws_id").strip()
+            item["header"] = f"\u2699 task_list add: {title[:60]}"
+            item["preview"] = f"status={status} child_ws_id={child_ws_id or '-'}"
+            item["title"] = title
+            item["status"] = status
+            item["child_ws_id"] = child_ws_id
+        elif action == "update":
+            task_id = self._coord_str_arg(args, "task_id").strip()
+            if not task_id:
+                return self._coord_tool_error(call_id, "task_list", "update: task_id is required")
+            # Reject non-string field values outright — avoids a
+            # preview/execute divergence where the approver sees
+            # ``title=42`` but the coercion below drops it to ``None`` and
+            # the mutation silently no-ops on that field.  Local names
+            # distinct from the ``add`` branch so mypy doesn't try to
+            # unify ``str`` and ``Any | None`` across mutually-exclusive
+            # branches.
+            upd_title: Any = args.get("title")
+            upd_status: Any = args.get("status")
+            upd_child: Any = args.get("child_ws_id")
+            for field_name, field_val in (
+                ("title", upd_title),
+                ("status", upd_status),
+                ("child_ws_id", upd_child),
+            ):
+                if field_val is not None and not isinstance(field_val, str):
+                    return self._coord_tool_error(
+                        call_id,
+                        "task_list",
+                        f"update: {field_name} must be a string",
+                    )
+            if upd_title is None and upd_status is None and upd_child is None:
+                return self._coord_tool_error(
+                    call_id,
+                    "task_list",
+                    "update: at least one of title / status / child_ws_id is required",
+                )
+            item["header"] = f"\u2699 task_list update: {task_id}"
+            bits: list[str] = []
+            if upd_title is not None:
+                bits.append(f"title={upd_title[:60]}")
+            if upd_status is not None:
+                bits.append(f"status={upd_status}")
+            if upd_child is not None:
+                bits.append(f"child_ws_id={upd_child or '-'}")
+            item["preview"] = " ".join(bits)
+            item["task_id"] = task_id
+            item["title"] = upd_title
+            item["status"] = upd_status
+            item["child_ws_id"] = upd_child
+        elif action == "remove":
+            task_id = self._coord_str_arg(args, "task_id").strip()
+            if not task_id:
+                return self._coord_tool_error(call_id, "task_list", "remove: task_id is required")
+            item["header"] = f"\u2699 task_list remove: {task_id}"
+            item["preview"] = ""
+            item["task_id"] = task_id
+        elif action == "reorder":
+            raw_ids = args.get("task_ids")
+            if not isinstance(raw_ids, list) or not all(isinstance(x, str) for x in raw_ids):
+                return self._coord_tool_error(
+                    call_id, "task_list", "reorder: task_ids must be a list of strings"
+                )
+            item["header"] = f"\u2699 task_list reorder: {len(raw_ids)} ids"
+            item["preview"] = ",".join(raw_ids[:6]) + ("..." if len(raw_ids) > 6 else "")
+            item["task_ids"] = raw_ids
+        return item
+
+    def _exec_task_list(self, item: dict[str, Any]) -> tuple[str, str]:
+        call_id = item["call_id"]
+        action = item["action"]
+        try:
+            if action == "list":
+                envelope = self._coord_client.task_list_get(self._ws_id)
+                tasks = envelope.get("tasks", [])
+                truncated = len(tasks) > 200
+                tasks = tasks[:200]
+                result: dict[str, Any] = {"tasks": tasks, "truncated": truncated}
+            elif action == "add":
+                result = self._coord_client.task_list_add(
+                    self._ws_id,
+                    title=item["title"],
+                    status=item["status"],
+                    child_ws_id=item["child_ws_id"],
+                )
+            elif action == "update":
+                result = self._coord_client.task_list_update(
+                    self._ws_id,
+                    task_id=item["task_id"],
+                    title=item["title"],
+                    status=item["status"],
+                    child_ws_id=item["child_ws_id"],
+                )
+            elif action == "remove":
+                result = self._coord_client.task_list_remove(self._ws_id, task_id=item["task_id"])
+            elif action == "reorder":
+                result = self._coord_client.task_list_reorder(
+                    self._ws_id, task_ids=item["task_ids"]
+                )
+            else:  # unreachable — _prepare validated the enum
+                result = {"error": f"unknown action: {action}"}
+        except Exception as e:
+            msg = f"Error: task_list {action} failed: {e}"
+            self._report_tool_result(call_id, "task_list", msg, is_error=True)
+            return call_id, msg
+        output = json.dumps(result, separators=(",", ":"), default=str)
+        if action == "list":
+            total = len(result.get("tasks", []))
+            summary = f"{total} tasks"
+            if result.get("truncated"):
+                summary += " (truncated at 200)"
+        elif "error" in result:
+            summary = f"{action} error: {result['error']}"
+        elif action == "add":
+            summary = f"added task {result.get('id', '?')}"
+        elif action == "update":
+            summary = f"updated task {result.get('id', item.get('task_id', '?'))}"
+        elif action == "remove":
+            summary = f"removed task {item.get('task_id', '?')}"
+        elif action == "reorder":
+            summary = f"reordered {len(item.get('task_ids', []))} tasks"
+        else:
+            summary = action
+        is_error = "error" in result
+        self._report_tool_result(call_id, "task_list", summary, is_error=is_error)
+        return call_id, self._truncate_output(output)
+
     def _prepare_memory(self, call_id: str, args: dict[str, Any]) -> dict[str, Any]:
         """Prepare a memory tool action (save/get/search/delete/list)."""
         action = (args.get("action") or "").strip().lower()

turnstone/core/storage/_postgresql.py

@@ -2423,6 +2423,41 @@ class PostgreSQLBackend:
                 _row_to_dict(r, "is_default", "readonly", "auto_approve", "enabled") for r in rows
             ]
 
+    def list_skills_filtered(
+        self,
+        *,
+        category: str | None = None,
+        tag: str | None = None,
+        scan_status: str | None = None,
+        enabled_only: bool = False,
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        with self._conn() as conn:
+            q = sa.select(prompt_templates).order_by(
+                prompt_templates.c.priority, prompt_templates.c.name
+            )
+            if category:
+                q = q.where(prompt_templates.c.category == category)
+            if scan_status:
+                q = q.where(prompt_templates.c.scan_status == scan_status)
+            if enabled_only:
+                q = q.where(prompt_templates.c.enabled == 1)
+            if tag:
+                # Quote-bracketed substring match against the JSON-array text:
+                # `["foo"]` matches `tag="foo"` but not `["foobar"]`.
+                # Escape ILIKE metacharacters (``%`` / ``_``) so a literal
+                # tag like ``a%b`` doesn't over-match.  ``ilike`` is
+                # case-insensitive natively on PostgreSQL — matches the
+                # SQLite backend's normalised behaviour.
+                pattern = f'%"{_escape_ilike(tag)}"%'
+                q = q.where(prompt_templates.c.tags.ilike(pattern, escape="\\"))
+            if limit > 0:
+                q = q.limit(limit)
+            rows = conn.execute(q).fetchall()
+            return [
+                _row_to_dict(r, "is_default", "readonly", "auto_approve", "enabled") for r in rows
+            ]
+
     def get_skill_by_name(self, name: str) -> dict[str, Any] | None:
         return self.get_prompt_template_by_name(name)
 

turnstone/core/storage/_protocol.py

@@ -917,6 +917,27 @@ class StorageBackend(Protocol):
         """Return prompt templates filtered by activation value, ordered by priority then name."""
         ...
 
+    def list_skills_filtered(
+        self,
+        *,
+        category: str | None = None,
+        tag: str | None = None,
+        scan_status: str | None = None,
+        enabled_only: bool = False,
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        """Return prompt templates filtered by optional category/tag/scan_status,
+        ordered by priority then name.
+
+        Filters are pushed into SQL — no per-row Python filter loops.  The
+        ``tag`` filter matches if the tag string appears in the JSON-array
+        ``tags`` column (quote-bracketed substring against the JSON text:
+        ``%"<tag>"%``).  Cheap and correct for tag values without quote
+        characters; upgrade to true JSON-array containment if the
+        convention ever needs to expand.
+        """
+        ...
+
     def get_skill_by_name(self, name: str) -> dict[str, Any] | None:
         """Lookup skill (prompt template) by name. Returns dict or None."""
         ...

turnstone/core/storage/_sqlite.py

@@ -2493,6 +2493,42 @@ class SQLiteBackend:
                 _row_to_dict(r, "is_default", "readonly", "auto_approve", "enabled") for r in rows
             ]
 
+    def list_skills_filtered(
+        self,
+        *,
+        category: str | None = None,
+        tag: str | None = None,
+        scan_status: str | None = None,
+        enabled_only: bool = False,
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        with self._conn() as conn:
+            q = sa.select(prompt_templates).order_by(
+                prompt_templates.c.priority, prompt_templates.c.name
+            )
+            if category:
+                q = q.where(prompt_templates.c.category == category)
+            if scan_status:
+                q = q.where(prompt_templates.c.scan_status == scan_status)
+            if enabled_only:
+                q = q.where(prompt_templates.c.enabled == 1)
+            if tag:
+                # Quote-bracketed substring match against the JSON-array text:
+                # `["foo"]` matches `tag="foo"` but not `["foobar"]`.
+                # Escape LIKE metacharacters (``%`` / ``_``) so a literal
+                # tag like ``a%b`` doesn't over-match, and lowercase both
+                # sides so the filter is case-insensitive on both backends
+                # (PostgreSQL ``LIKE`` is case-sensitive; SQLite is not —
+                # normalising here prevents dev/prod divergence).
+                pattern = f'%"{_escape_like(tag.lower())}"%'
+                q = q.where(sa.func.lower(prompt_templates.c.tags).like(pattern, escape="\\"))
+            if limit > 0:
+                q = q.limit(limit)
+            rows = conn.execute(q).fetchall()
+            return [
+                _row_to_dict(r, "is_default", "readonly", "auto_approve", "enabled") for r in rows
+            ]
+
     def get_skill_by_name(self, name: str) -> dict[str, Any] | None:
         return self.get_prompt_template_by_name(name)
 

turnstone/tools/list_nodes.json

@@ -0,0 +1,22 @@
+{
+  "name": "list_nodes",
+  "description": "List server nodes in the cluster with their metadata. Auto-approved — safe, read-only. Each node carries two metadata sources: (a) AUTO-populated on every node startup — keys arch, cpu_count, fqdn, hostname, os, os_release, python (always present); (b) USER-supplied via the console Nodes admin tab — keys like capability, region, tenant, role (deployment-specific). Pass arbitrary key=value filters to narrow the result; ALL filters must match (AND semantics). Pair with target_node on spawn_workstream to pin a child workstream to a node that matches your capability requirements.",
+  "parameters": {
+    "type": "object",
+    "properties": {
+      "filters": {
+        "type": "object",
+        "description": "Optional dict of key=value pairs to filter by. Pass values in their natural JSON type — string, integer, number, or boolean. Filter comparison is JSON-equal against the stored value (e.g. cpu_count=4 as int matches the stored integer, NOT the string \"4\"). Omit to list all nodes. Examples: {\"arch\": \"x86_64\"}, {\"capability\": \"gpu\", \"region\": \"us-west\"}, {\"cpu_count\": 4}.",
+        "additionalProperties": {
+          "type": ["string", "integer", "number", "boolean"]
+        }
+      },
+      "limit": {
+        "type": "integer",
+        "description": "Max rows to return. Default 100, max 500."
+      }
+    }
+  },
+  "coordinator": true,
+  "auto_approve": true
+}

turnstone/tools/list_skills.json

@@ -0,0 +1,31 @@
+{
+  "name": "list_skills",
+  "description": "List skills (worker profiles) available in the cluster. Auto-approved — safe, read-only. Pass optional filters to narrow by category (e.g. 'engineering', 'ops'), tag (single tag from the skill's tags array), or scan_status ('clean', 'flagged', 'unscanned', 'pending'). Use to discover which skill names you can pass to spawn_workstream. Returns name, category, tags, version, description, model preference, enabled flag, scan_status, activation — enough for an informed pick.",
+  "parameters": {
+    "type": "object",
+    "properties": {
+      "category": {
+        "type": "string",
+        "description": "Filter by category (exact match)."
+      },
+      "tag": {
+        "type": "string",
+        "description": "Filter by tag (must appear in the skill's tags array)."
+      },
+      "scan_status": {
+        "type": "string",
+        "description": "Filter by scan status: clean / flagged / unscanned / pending."
+      },
+      "enabled_only": {
+        "type": "boolean",
+        "description": "If true, return only enabled skills. Default false."
+      },
+      "limit": {
+        "type": "integer",
+        "description": "Max rows to return. Default 100, max 500."
+      }
+    }
+  },
+  "coordinator": true,
+  "auto_approve": true
+}

turnstone/tools/task_list.json

@@ -0,0 +1,39 @@
+{
+  "name": "task_list",
+  "description": "Lightweight, ordered task list for the coordinator's own planning state. Persisted on the coordinator workstream so it survives restarts. Use to track work decomposition, mark progress, and link tasks to spawned children (child_ws_id field). Five actions: add (append a task, needs approval), update (mutate title/status/child_ws_id by id, needs approval), remove (delete by id, needs approval), reorder (rearrange by id list, needs approval), list (read, auto-approved). Status enum: pending / in_progress / done / blocked.",
+  "parameters": {
+    "type": "object",
+    "properties": {
+      "action": {
+        "type": "string",
+        "description": "One of: add / update / remove / reorder / list.",
+        "enum": ["add", "update", "remove", "reorder", "list"]
+      },
+      "title": {
+        "type": "string",
+        "description": "For 'add' and 'update': task title (max 200 chars)."
+      },
+      "task_id": {
+        "type": "string",
+        "description": "For 'update' and 'remove': target task id."
+      },
+      "status": {
+        "type": "string",
+        "description": "For 'add' and 'update': pending / in_progress / done / blocked.",
+        "enum": ["pending", "in_progress", "done", "blocked"]
+      },
+      "child_ws_id": {
+        "type": "string",
+        "description": "For 'add' and 'update': optional ws_id of the child workstream this task represents."
+      },
+      "task_ids": {
+        "type": "array",
+        "items": { "type": "string" },
+        "description": "For 'reorder': new ordering by task_id (must be a permutation of existing task_ids)."
+      }
+    },
+    "required": ["action"]
+  },
+  "coordinator": true,
+  "primary_key": "action"
+}