dongarandClaude Opus 5.5
1ad0b2147d
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 1m22s
Development Plan M0.1. Every section reference in code, tests, specs/, the example config, the tuning guide, and the CLAUDE.md and README files now cites docs/Architecture.md (P, L, D, O, X, Q) instead of the old spec's A-H, and docs/Architecture-Build-Spec.md is deleted (git keeps it). The plan's old-to-new table goes with it. Mostly mechanical, with a per-site choice where one old section split: A4 became L2 where it meant stop/abandon/the loop and L3 where it meant approvals; A10 became L6 in the context-budget code and L7 elsewhere; B2 became D4 for outbound Gitea and publishing and D5 for review intake; 'A3 step N' became the matching step of L8's boot sequence. Comments describing behaviour the architecture drops (planning always escalating, thresholds tuned at an old build step) now describe the current code and point at the Development Plan. docs/Architecture.md gains one line the rewrite missed: intake outages delay but never fail work (old B2), now in L5, where the poller and Matrix code cite it. Comments and docs only; no behaviour change. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
139 lines
4.5 KiB
Python
139 lines
4.5 KiB
Python
"""The one thread that runs orchestrator work and owns the state store.
|
|
|
|
Planning turns, milestones and review responses can each run for minutes.
|
|
Running them on the thread that delivered the message blocked that thread for
|
|
as long: the Matrix sync loop read nothing new until the work finished, so an
|
|
operator's `stop` (L2) would only have arrived after the work it was meant to
|
|
stop. Every handler now runs here instead, and transports only hand work over.
|
|
|
|
Owning the store is the other half. SQLite connections are bound to the thread
|
|
that opened them, and the store keeps that check on: it is opened on this
|
|
thread, used only from it, and any future code that touches it from elsewhere
|
|
fails loudly instead of racing.
|
|
|
|
Jobs run one at a time, in order, so state changes never interleave — the
|
|
guarantee the orchestrator's lock used to give.
|
|
"""
|
|
|
|
import sys
|
|
import threading
|
|
import traceback
|
|
from collections import deque
|
|
from collections.abc import Callable
|
|
from concurrent.futures import Future
|
|
from queue import Queue
|
|
from typing import Protocol, TypeVar
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class Worker(Protocol):
|
|
@property
|
|
def busy(self) -> bool: ...
|
|
|
|
def submit(self, job: Callable[[], object]) -> None: ...
|
|
|
|
def call(self, job: Callable[[], T], timeout: float | None = None) -> T: ...
|
|
|
|
|
|
class InlineWorker:
|
|
"""Runs jobs on the caller's thread, for tests and single-threaded use.
|
|
|
|
A job submitted from inside a running job waits until that one finishes,
|
|
exactly as it would queue behind it on the real worker. Without that, a
|
|
test that says `stop` from inside a fake milestone would run the stop
|
|
handling in the middle of the milestone, which production never does.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._pending: deque[Callable[[], object]] = deque()
|
|
self._running = False
|
|
|
|
@property
|
|
def busy(self) -> bool:
|
|
return self._running
|
|
|
|
def submit(self, job: Callable[[], object]) -> None:
|
|
self._pending.append(job)
|
|
if self._running:
|
|
return
|
|
self._running = True
|
|
try:
|
|
while self._pending:
|
|
self._pending.popleft()()
|
|
finally:
|
|
self._running = False
|
|
|
|
def call(self, job: Callable[[], T], timeout: float | None = None) -> T:
|
|
if self._running:
|
|
return job() # already on the worker, as a nested call would be
|
|
box: list[T] = []
|
|
self.submit(lambda: box.append(job()))
|
|
return box[0]
|
|
|
|
|
|
_STOP = object()
|
|
|
|
|
|
def _report(exc: BaseException) -> None:
|
|
traceback.print_exception(exc, file=sys.stderr)
|
|
|
|
|
|
class SerialWorker:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
on_error: Callable[[BaseException], None] = _report,
|
|
name: str = "nisse-worker",
|
|
) -> None:
|
|
self._queue: Queue[object] = Queue()
|
|
self._on_error = on_error
|
|
self._busy = False
|
|
self._thread = threading.Thread(target=self._run, name=name, daemon=True)
|
|
|
|
@property
|
|
def busy(self) -> bool:
|
|
return self._busy
|
|
|
|
def start(self) -> None:
|
|
self._thread.start()
|
|
|
|
def submit(self, job: Callable[[], object]) -> None:
|
|
self._queue.put((job, None))
|
|
|
|
def call(self, job: Callable[[], T], timeout: float | None = None) -> T:
|
|
"""Run a job here and wait for its result. Raises TimeoutError if it
|
|
has not finished in time; the job itself still runs."""
|
|
if threading.current_thread() is self._thread:
|
|
return job()
|
|
future: Future[T] = Future()
|
|
self._queue.put((job, future))
|
|
return future.result(timeout)
|
|
|
|
def stop(self, timeout: float | None = None) -> None:
|
|
self._queue.put(_STOP)
|
|
self._thread.join(timeout)
|
|
|
|
def _run(self) -> None:
|
|
while True:
|
|
item = self._queue.get()
|
|
if item is _STOP:
|
|
return
|
|
job, future = item # type: ignore[misc]
|
|
self._busy = True
|
|
try:
|
|
result = job()
|
|
except Exception as exc: # noqa: BLE001 -- see below
|
|
# One failed job must not take the agent's only worker with
|
|
# it. A caller that waits gets the exception; otherwise it is
|
|
# reported and the next job runs.
|
|
if future is not None:
|
|
future.set_exception(exc)
|
|
else:
|
|
self._on_error(exc)
|
|
else:
|
|
if future is not None:
|
|
future.set_result(result)
|
|
finally:
|
|
self._busy = False
|