mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-27 14:24:47 -06:00
Patrick Buckley
94dcaf34fd
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).
(cherry picked from commit e60c19befd)
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
|