mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
Patrick Buckley
e60c19befd
Wake path: - Denial metacog nudge moves to the tool channel so it drains with the denied tool batch instead of the next user-message seam. - wake_workstream_if_pending: shared wake gate for watch fires on already-idle workstreams (no IDLE transition for the watcher to observe), wired as wake_fn at every set_watch_runner site via the shared _watch_fire_wake_fn helper (closes over the Workstream OBJECT — after eviction+restore an id-keyed manager lookup would miss). - session_worker exit backstop re-runs the wake gate the moment worker ownership clears: IDLE fans out on the worker thread, so transition-time wakes always landed on the reuse path and no-op'd (the coordinator idle_children strand). - deliver_wake_nudge_from_queue contains GenerationCancelled — it is the wake worker's run() closure and only Exception is caught downstream. Watch delivery: - Terminal fires that cannot reach their workstream are HELD and redelivered on min(interval, 60s) without re-running the command, bounded by MAX_DELIVERY_ATTEMPTS per cycle and the watch's own max_polls across cycles; the poll charge commits durably at hold time so restarts stay budget-bounded. - Restore admission control: per-ws dedup + MAX_CONCURRENT_RESTORES cap, presence-only re-check under the lock, detection-only stall alerts (reclaiming a wedged admission would trade capped degradation for total poll-pool collapse). - Permanent-vs-transient restore taxonomy: corrupt persona stamp and genuinely-missing history (confirmed by a raising storage probe — the resume loader swallows read blips into []) deactivate the watch immediately; everything else holds and retries. - Cancel-race defense: delivery paths re-check is_watch_active before stashing/dispatching, cancel paths write the row BEFORE forget_terminal_dispatched, the HTTP cancel endpoint clears runner state, and a per-tick sweep bounds the residual stash-after-clear interleaving to one check_interval. - Abandon/exhaustion commits are write-then-clear so storage that can read but not write retries the row write instead of re-running the command every cycle; the fresh-fire unrestorable path stashes before its deactivation write for the same reason. Registry follows identity: - The dispatch registry is keyed by _ws_id at registration time; every rebind now moves it: non-fork resume() and /new go through _follow_watch_registration (new key live before the old is removed, never stealing a registration another live session holds), removals are owner-checked so tearing down a watch-restore shell or a resumed-away session cannot unregister a live pane, the restore shell yields to a registration that appears mid-restore, CLI --resume registers after the successful resume, and both the open path and the detail-GET lazy rehydrate wire the registration. Teardown gating and backpressure honesty: - cleanup_session_ui marks ws._closed FIRST under ws._lock — every teardown path (close, close_idle, evict, delete, discard) funnels through it — and session_worker.send re-checks under the same lock, so a wake can never spawn a worker on a torn-down workstream. - Create responses carry initial_message_status when the initial message could not be delivered (queue_full / refused_closed) instead of reading as success; staged attachments survive for the retry; /send surfaces a closed workstream as 404 rather than queue_full. Docs/spec: OpenAPI artifacts regenerated; api-reference documents the new create-response field; TS SDK type extended. Tests: ~30 new pins (cancel races, budget durability across restarts, owner-checked registry moves, teardown gating, stall alerts, backpressure surfaces, wait_until final re-check); wide subsystem sweep green (2353 passed).
82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
"""Shared test helpers — kept out of conftest.py since these are factories,
|
|
not fixtures, and several test files want to import them directly."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import TYPE_CHECKING, Any
|
|
from unittest.mock import MagicMock
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Callable
|
|
|
|
|
|
def wait_until(cond: Callable[[], bool], timeout: float = 5.0) -> None:
|
|
"""Poll ``cond`` to True within ``timeout`` or fail the test.
|
|
|
|
The worker/wake tests can't join threads by identity:
|
|
``session_worker.send`` assigns ``ws.worker_thread`` under the lock
|
|
BEFORE ``t.start()``, so the instant a dispatching call returns, a
|
|
fast worker may already have run its exit backstop and installed the
|
|
(not-yet-started) wake thread — joining whatever ``ws.worker_thread``
|
|
points at races ``RuntimeError: cannot join thread before it is
|
|
started``. Poll outcomes instead.
|
|
"""
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
if cond():
|
|
return
|
|
time.sleep(0.005)
|
|
if cond():
|
|
# Final re-check: the condition can become true during the last
|
|
# sleep (or a CI descheduling stall past the deadline) — failing
|
|
# without re-looking makes the helper itself a flake source.
|
|
return
|
|
raise AssertionError("condition not met within timeout")
|
|
|
|
|
|
def make_chat_session(**overrides: Any) -> Any:
|
|
"""Build a minimal ``ChatSession`` with sane test defaults.
|
|
|
|
Caller passes any constructor arg as a kwarg to override the default —
|
|
e.g. ``make_chat_session(memory_config=MemoryConfig(fetch_limit=5))``.
|
|
"""
|
|
from turnstone.core.session import ChatSession
|
|
|
|
defaults: dict[str, Any] = {
|
|
"client": MagicMock(),
|
|
"model": "test-model",
|
|
"ui": MagicMock(),
|
|
"instructions": None,
|
|
"temperature": 0.5,
|
|
"max_tokens": 4096,
|
|
"tool_timeout": 30,
|
|
}
|
|
defaults.update(overrides)
|
|
return ChatSession(**defaults)
|
|
|
|
|
|
def patch_session_storage(
|
|
monkeypatch: Any,
|
|
*,
|
|
active: bool = True,
|
|
raise_on_is_active: bool = False,
|
|
) -> list[str]:
|
|
"""Patch ``session.get_storage`` to a stub whose ``is_watch_active``
|
|
returns *active* (or raises if *raise_on_is_active*). Returns the
|
|
list of ``watch_id``s the predicate was called with.
|
|
"""
|
|
from turnstone.core import session as session_mod
|
|
|
|
calls: list[str] = []
|
|
|
|
class _Stub:
|
|
def is_watch_active(self, watch_id: str) -> bool:
|
|
calls.append(watch_id)
|
|
if raise_on_is_active:
|
|
raise RuntimeError("storage down")
|
|
return active
|
|
|
|
monkeypatch.setattr(session_mod, "get_storage", lambda: _Stub())
|
|
return calls
|