"""Scaffold core loop (v2). run_scaffold(...) loads the author's entrypoint, finds and validates scan(), builds the ScanContext, then drives the interval loop: time-box per tick, graceful stop on SIGTERM/SIGINT, parent-death detection via ppid, transactional state via ctx.state, candidates to sink. Each tick mints one id, handed to the sink with the tick's candidates and carried on every error the tick reports. `path` is a scaffold concern only — it is put on sys.path before import and never passed to scan(). scan() is called as scan(inputs, ctx) — 2-arg, sync, returning a plain list[dict]. State mutates ONLY via ctx.state (commit on a clean tick, rollback on error/timeout/malformed/failed-persist — transactional discard wins on failure). A tick is also NOT clean when every ctx.senpi_mcp call in it failed and scan() returned no candidates (`mcp_error`): authored code routinely swallows the MCP exception and returns [], which would otherwise report a blind scanner as ok forever. The client tallies its failures per tick; the loop reads the tally after scan() returns and reclassifies. MCP failures alongside candidates stay ok, with the count carried on `scaffold_tick_finished` as `mcp_error_count`. """ import inspect import math import os import time import threading from typing import Any, Callable, Optional from .ctx import ScanContext from .entrypoint_loader import ( ScanEntrypointError, ScanSignatureError, import_entrypoint, locate_scan, verify_scan_signature, ) from .loop_primitives import ( _TickTimeout, _arm_tick_alarm, _disarm_tick_alarm, _install_shutdown_handlers, _interruptible_sleep, log_event, ) from .state_manager import StateManager from .tick_facts import build_tick_facts, count_changed, project_state_row from .ulid import new_ulid def _read_call_timeout_seconds(senpi_mcp_mod: Any) -> float: """The per-call MCP bound from the launch env, or the documented default. `SENPI_MCP_CALL_TIMEOUT_SECONDS` is operator-set on the runtime box and reaches the scaffold child through the supervisor's inherited env. A value that is not a positive finite number must not take the scanner down at launch — it falls back to the default and says so (`senpi_mcp_config_invalid`), the same "loud, not fatal" shape as a malformed launch config field. """ variable = senpi_mcp_mod.CALL_TIMEOUT_ENV_VAR default = senpi_mcp_mod.DEFAULT_CALL_TIMEOUT_SECONDS raw = os.environ.get(variable) if raw is None: return default try: seconds = float(raw) except ValueError: seconds = math.nan if not senpi_mcp_mod.is_valid_call_timeout(seconds): log_event( "senpi_mcp_config_invalid", variable=variable, value=raw[:100], fallback_seconds=default, ) return default return seconds def _build_senpi_mcp() -> Any: """Construct the MCP client when creds are present; else return None. Creds come from the launch env (SENPI_API_KEY / SENPI_MCP_URL), injected by the launcher regardless of how the operator supplied them. With neither set, return None so the no-creds path does not crash — the client is loud at FIRST USE (see scaffold.senpi_mcp). Reference the class through the module attribute so a test stub patched onto scaffold.senpi_mcp is honored. The per-call bound comes from SENPI_MCP_CALL_TIMEOUT_SECONDS (see _read_call_timeout_seconds). """ api_key = os.environ.get("SENPI_AUTH_TOKEN") or os.environ.get("SENPI_API_KEY") or None mcp_url = os.environ.get("SENPI_MCP_URL") or None if not api_key and not mcp_url: return None from . import senpi_mcp as senpi_mcp_mod # Read-only-by-default producer MCP boundary: the set of MUTATION tools # producers are permitted to call is owned SOLELY by the scaffold-source # PRODUCER_WRITE_ALLOWLIST constant (EMPTY by default → all mutations # blocked). It is edited by the scaffold maintainer in senpi_mcp.py — it is # NOT operator/env-configurable and NOT reachable by authored producer code. return senpi_mcp_mod.SenpiMcpClient( api_key=api_key, mcp_url=mcp_url, write_allowlist=senpi_mcp_mod.PRODUCER_WRITE_ALLOWLIST, call_timeout_seconds=_read_call_timeout_seconds(senpi_mcp_mod), ) def _accepts_kwarg(fn: Any, name: str) -> bool: """Whether `fn` takes `name` as a keyword argument. Read from the signature, once, rather than calling with the keyword and retrying on TypeError: a sink that raises TypeError from its own body would then be called twice and deliver the same candidates twice. A callable whose signature cannot be read keeps the narrower call it has today. """ try: params = list(inspect.signature(fn).parameters.values()) except (TypeError, ValueError): return False return any( p.kind == inspect.Parameter.VAR_KEYWORD or ( p.name == name and p.kind in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY) ) for p in params ) def _call_scan(scan: Callable, inputs: dict, ctx: Any): """Call scan(inputs, ctx) — the v2 2-arg contract. There is no `client` and no 3-arg path. scan returns a plain list[dict]; validation of that return happens in the tick loop. """ return scan(inputs, ctx) def run_scaffold( path: str, entrypoint: str, *, interval: float, timeout_seconds: Optional[float] = None, inputs: Optional[dict] = None, state_path: Optional[str] = None, scanner_name: str, wallet: str, default_signal_validity_seconds: int, state_history_max_count: int = 0, sink: Optional[Callable] = None, max_ticks: Optional[int] = None, install_signal_handlers: bool = True, senpi_mcp: Optional[Any] = None, dry_run: bool = False, ) -> int: """Load the author's entrypoint and run its scan() on an interval loop (v2). Args: path: Directory added to sys.path before import. Never passed to scan(). entrypoint: Module name to import from `path`. interval: Seconds between tick starts. Must be > 0. timeout_seconds: Per-tick wall-clock budget. Defaults to `interval`. Must be > 0. Enforced via SIGALRM on platforms that support it. inputs: Tunables dict passed as scan()'s first argument. Default {}. state_path: Path to the JSON state file. None means no persistence. scanner_name: This scanner's recipe name (read-only ctx metadata). wallet: Strategy wallet (read-only ctx metadata). default_signal_validity_seconds: Fallback signal TTL (seconds); threaded to the envelope path. state_history_max_count: Bound on the persisted state series. 0/unset = history disabled (ctx.state.append raises). sink: Callable(candidates, tick_id=None) invoked after each clean tick. A sink that takes candidates only is called with candidates only. Default no-op. max_ticks: Stop after this many ticks. None = unbounded. Used by tests and by a validation run, which forces exactly one tick. senpi_mcp: Data client to hand to scan(). Default: built from the launch environment. Injected by a validation run so reads can be observed. dry_run: Sets ctx.dry_run. True only during validation. install_signal_handlers: Install SIGTERM/SIGINT graceful-stop handlers. False for tests. Returns: Total ticks attempted. Raises: ValueError: interval <= 0 or timeout_seconds <= 0. ScanEntrypointError: import failure, scan missing or not callable. ScanSignatureError: scan call-shape incompatible with scaffold contract. """ # ---------- Launch validation (raises before any tick) ---------- if interval <= 0: raise ValueError(f"interval must be > 0; got {interval!r}") if timeout_seconds is None: timeout_seconds = interval if timeout_seconds <= 0: raise ValueError(f"timeout_seconds must be > 0; got {timeout_seconds!r}") if inputs is None: inputs = {} # Import and locate scan — any failure raises before tick 1. module = import_entrypoint(path, entrypoint) scan = locate_scan(module) verify_scan_signature(scan) # ---------- ctx + state setup ---------- # The StateManager (B2) is the persisted series. It reads any prior state from # disk on construction. state_history_max_count 0/unset => history disabled # (append raises). It is passed into ctx AS-IS (never wrapped). state = StateManager(state_path, state_history_max_count) if state_path is not None \ else StateManager(None, state_history_max_count) # ctx.interval_seconds is the integer-seconds cadence (design §2.1). The loop's # `interval` may be a sub-second float in tests; expose the int seconds when it # is whole, else the float as-is. interval_seconds = int(interval) if float(interval).is_integer() else interval # B5: fill ctx.senpi_mcp when MCP creds are present (resolved at launch from # SENPI_API_KEY / SENPI_MCP_URL). With no creds the slot stays None and launch # must NOT crash — the client is loud at FIRST USE, not at construction. Import # is deferred so a stub patched onto scaffold.senpi_mcp in tests is honored. # An injected client wins, so a validation run can observe what the tick reads without # changing how it reads it. Absent one this behaves exactly as before. senpi_mcp = senpi_mcp if senpi_mcp is not None else _build_senpi_mcp() ctx = ScanContext( state=state, wallet=wallet, scanner_name=scanner_name, interval_seconds=interval_seconds, senpi_mcp=senpi_mcp, dry_run=dry_run, ) # ---------- Lifecycle setup ---------- stop_event = threading.Event() if install_signal_handlers: _install_shutdown_handlers(stop_event) start_ppid = os.getppid() log_event( "scaffold_started", path=path, entrypoint=entrypoint, interval=interval, timeout_seconds=timeout_seconds, scanner_name=scanner_name, max_ticks=max_ticks, ) tick_count = 0 # The MCP boundary's per-tick failure tally, when the client exposes one. Duck-typed: # an injected stub without it (a validation run) simply never reclassifies a tick. mcp_tally = getattr(senpi_mcp, "tick_errors", None) sink_takes_tick_id = sink is not None and _accepts_kwarg(sink, "tick_id") # The per-tick observability facts ride the POST the sink already makes. A sink that # predates them is called exactly as before, so an older sink is not broken by them. sink_takes_tick = sink is not None and _accepts_kwarg(sink, "tick") post_error_takes_tick = ( sink is not None and hasattr(sink, "post_error") and _accepts_kwarg(sink.post_error, "tick") ) # Consecutive ticks that produced no candidate. 47 quiet ticks is information; the word # "quiet" repeated 47 times is not, so the streak is counted here and sent as a number. quiet_ticks = 0 # The previous tick's start, for the cadence lag. None on the first tick of a process: # there is no gap to measure yet, and a 0 would claim the scanner was exactly on time. prev_tick_started_at: Optional[float] = None # The last projection reported, so `state_changed` says what MOVED rather than what exists. prev_state_projection: Optional[dict] = None try: while not stop_event.is_set(): # --- Parent-death check (before tick) --- current_ppid = os.getppid() if current_ppid != start_ppid or current_ppid == 1: log_event("scaffold_parent_gone", start_ppid=start_ppid, current_ppid=current_ppid) break tick_count += 1 # One id per tick, minted before any of the tick's work so every candidate and every # error it reports names the same tick. tick_id = new_ulid() tick_started_at = time.time() # Actual gap between tick starts, minus the cadence the scanner is configured for. # A positive number is a scanner falling behind its own interval. lag_ms = ( None if prev_tick_started_at is None else int(round((tick_started_at - prev_tick_started_at - interval) * 1000)) ) prev_tick_started_at = tick_started_at tick_status = "ok" tick_error = None candidates: list = [] mcp_error_count = 0 if mcp_tally is not None: mcp_tally.reset() armed = False try: armed = _arm_tick_alarm(timeout_seconds) try: result = _call_scan(scan, inputs, ctx) # v2 return contract: a PLAIN list[dict]. Anything else is # malformed (a tuple — the old v1 shape — included). if not isinstance(result, list): tick_status = "malformed" malformed_reason = ( f"scan() returned {type(result).__name__!r}; expected a " f"plain list[dict] (the v1 (candidates, state) tuple is gone)" ) # Carried on the error record, not only in this log line: the sink is what # a caller reads, and "malformed" without the reason is not actionable. tick_error = { "type": "malformed", "tick_id": tick_id, "message": malformed_reason, } log_event( "scaffold_tick_malformed", tick=tick_count, tick_id=tick_id, reason=malformed_reason, ) else: candidates = result except _TickTimeout as exc: tick_status = "timeout" candidates = [] tick_error = {"type": "timeout", "tick_id": tick_id, "message": str(exc)} log_event( "scaffold_tick_timeout", tick=tick_count, tick_id=tick_id, reason=str(exc), ) except Exception as exc: # noqa: BLE001 — author bug must not crash scaffold tick_status = "error" candidates = [] tick_error = { "type": "error", "tick_id": tick_id, "error_type": type(exc).__name__, "message": str(exc), } log_event( "scaffold_tick_error", tick=tick_count, tick_id=tick_id, error_type=type(exc).__name__, error=str(exc), ) finally: if armed: _disarm_tick_alarm() duration_ms = int((time.time() - tick_started_at) * 1000) # Authored code routinely catches the MCP exception and returns [], so a # scanner whose every tool call failed would otherwise finish `ok` with zero # candidates — and keep refreshing liveness while blind. Read the boundary's # own tally instead of trusting the return value: on an otherwise-ok tick, # MCP errors + no candidates is a failed tick; MCP errors + candidates stays # ok (a partial read still produced work). The count rides on the finished # event whatever the status, so an error/timeout tick shows its MCP side too. mcp_error_count, mcp_failure = _classify_mcp_errors(mcp_tally, candidates) if tick_status == "ok" and mcp_failure is not None: tick_status = "mcp_error" # Stamped here, not in the classifier: every error record posted to # `/errors` names the tick it came from, and the classifier is tick-agnostic. tick_error = {**mcp_failure, "tick_id": tick_id} log_event( "scaffold_tick_mcp_error", tick=tick_count, tick_id=tick_id, error_type=mcp_failure["error_type"], error_count=mcp_error_count, ) # The quiet streak counts ticks, not words: a tick that produced a # candidate ends the streak whatever its status. quiet_ticks = 0 if candidates else quiet_ticks + 1 # --- Clean tick: commit ctx.state, deliver candidates --- persist_error = None if tick_status == "ok": # Transactional commit. A failed persist RAISES (StateManager # enforces it) and does NOT advance the committed snapshot — so # the next tick recomputes from the un-advanced state. On a # commit failure we roll back and treat the tick as failed # (discard wins on failed-persist; no advance-on-persist-failure). committed = True try: ctx.state.commit() except Exception as commit_exc: # noqa: BLE001 committed = False ctx.state.rollback() log_event( "scaffold_state_persist_failed", tick=tick_count, tick_id=tick_id, error_type=type(commit_exc).__name__, error=str(commit_exc), ) persist_error = { "type": "state_persist_failed", "tick": tick_count, "tick_id": tick_id, "message": ( "scanner state could not be persisted; " "tick discarded, state not advanced" ), } commit_outcome = "committed" if committed else "persist_failed" else: # Non-ok tick (error / timeout / malformed): roll back the # working buffer (discard the tick's mutations — state not # advanced) and report to /errors via the sink, when supported. ctx.state.rollback() committed = False commit_outcome = "rolled_back" # Read AFTER the commit/rollback above, so this is always the last # COMMITTED row — stale by one tick on a failed tick, which is exactly # when the last state the scanner agreed on is worth seeing. state_row = ctx.state.last() state_projection = project_state_row(state_row) state_fields = len(state_row) if isinstance(state_row, dict) else None state_changed = ( None if state_fields is None else count_changed(state_projection, prev_state_projection) ) prev_state_projection = state_projection if state_fields is not None else None tick_facts = build_tick_facts( tick_id=tick_id, status=tick_status, candidate_count=len(candidates), duration_ms=duration_ms, commit=commit_outcome, quiet_ticks=quiet_ticks, lag_ms=lag_ms, mcp_snapshot=mcp_tally.snapshot() if mcp_tally is not None else None, call_snapshot=( mcp_tally.call_snapshot() if mcp_tally is not None and hasattr(mcp_tally, "call_snapshot") else None ), state_row=state_row, state_fields=state_fields, state_changed=state_changed, ) # Exactly one POST carries this tick's facts: /signals on a clean tick, # /errors on any tick that failed. Both already happen every tick, so the # facts add no request of their own. sink_can_post_error = sink is not None and hasattr(sink, "post_error") facts_for_error = tick_facts if post_error_takes_tick else None if tick_status == "ok": if persist_error is not None and sink_can_post_error: _post_error_safe( sink, persist_error, tick_count, tick_id, facts_for_error ) if committed and sink is not None: try: kwargs = {} if sink_takes_tick_id: kwargs["tick_id"] = tick_id if sink_takes_tick: kwargs["tick"] = tick_facts sink(candidates, **kwargs) except Exception as sink_exc: # noqa: BLE001 log_event( "scaffold_sink_error", tick=tick_count, tick_id=tick_id, error_type=type(sink_exc).__name__, error=str(sink_exc), ) elif sink_can_post_error: _post_error_safe( sink, tick_error or {"type": tick_status, "tick_id": tick_id}, tick_count, tick_id, facts_for_error, ) log_event( "scaffold_tick_finished", tick=tick_count, tick_id=tick_id, status=tick_status, duration_ms=duration_ms, candidate_count=len(candidates), # Present only when the tick saw MCP failures — an absent key is the # honest "none", a 0 every tick would be noise on the fingerprint. **({"mcp_error_count": mcp_error_count} if mcp_error_count else {}), ) except _TickTimeout as late_exc: # A SIGALRM queued by the OS, delivered after the inner try/except # exited (e.g. during commit, sink, or log_event). Absorb it: the # tick already did its work; disarm, roll back any uncommitted # mutations, log the late arrival, and let the loop continue. _disarm_tick_alarm() try: ctx.state.rollback() except Exception: # noqa: BLE001 pass log_event( "scaffold_tick_timeout_late", tick=tick_count, tick_id=tick_id, reason=str(late_exc), ) # --- max_ticks bound --- if max_ticks is not None and tick_count >= max_ticks: break # --- Parent-death check (after tick) --- current_ppid = os.getppid() if current_ppid != start_ppid or current_ppid == 1: log_event("scaffold_parent_gone", start_ppid=start_ppid, current_ppid=current_ppid) break # --- Inter-tick sleep --- elapsed = time.time() - tick_started_at sleep_for = max(0.0, interval - elapsed) if sleep_for > 0 and not stop_event.is_set(): _interruptible_sleep(sleep_for, stop_event) finally: log_event("scaffold_stopping", tick_count=tick_count) return tick_count def _classify_mcp_errors(mcp_tally: Any, candidates: list) -> tuple[int, Optional[dict]]: """Read the tick's MCP failure tally and decide whether the tick was really ok. Returns (mcp_error_count, tick_error). tick_error is an `mcp_error` record — dominant error type and count — ONLY when the tick had MCP failures and produced no candidates; a tick with failures that still produced candidates is ok and carries only the count. No tally (no client, or a stub without one) is clean. """ if mcp_tally is None: return 0, None snapshot = mcp_tally.snapshot() count = int(snapshot.get("count") or 0) if count == 0 or candidates: return count, None dominant = snapshot.get("dominant_error_type") or "unknown" plural = "" if count == 1 else "s" return count, { "type": "mcp_error", "error_type": dominant, "error_count": count, "message": ( f"{count} MCP tool call{plural} failed this tick (dominant {dominant}) " "and scan() produced no candidates; tick discarded" ), } def _post_error_safe( sink: Any, error: dict, tick_count: int, tick_id: str, tick: Optional[dict] = None, ) -> None: """Call sink.post_error(error), swallowing/logging any failure. `tick` is passed only when the sink's signature accepts it — the caller checks once at launch, so a sink that predates the per-tick facts keeps its one-argument call rather than being handed a keyword it would raise on. """ try: if tick is None: sink.post_error(error) else: sink.post_error(error, tick=tick) except Exception as perr: # noqa: BLE001 log_event( "scaffold_post_error_failed", tick=tick_count, tick_id=tick_id, error_type=type(perr).__name__, error=str(perr), )