"""Process entry for the scaffold core (v2). Usage: python3 -m scaffold The launch config JSON is the FROZEN launcher<->scaffold contract (design §4a). This side READS every §4a key the B1 launcher writes and threads them into run_scaffold. Hard cut: old key names are gone (`interval` -> `interval_seconds`, `params` -> `inputs`). Required keys (missing => fail loud, non-zero exit): path — directory added to sys.path (scanner working dir). entrypoint — module name to import from `path`. interval_seconds — tick cadence in seconds (was `interval`). default_signal_validity_seconds — fallback signal TTL (seconds); no magic default. wallet — strategy wallet (-> ctx.wallet). scanner_name — this scanner's recipe name (-> ctx.scanner_name). Optional keys: timeout_seconds — per-tick wall-clock budget (default: interval_seconds). state_path — JSON state file path (None => no persistence). inputs — tunables dict for scan() (was `params`); default {}. state_history_max_count — state series bound; default 0 = history disabled. max_ticks — stop after N ticks; absent = run until stopped. A validation run sets 1; production leaves it absent. Delivery sink is wired when EXTERNAL_SCANNER_ID and SIGNAL_INGESTION_URL are set in the environment. If either is absent the scaffold runs with the default no-op sink and logs a warning — scanners can still operate without a live runtime. """ import json import os import sys import traceback def _main(argv=None) -> int: if argv is None: argv = sys.argv if len(argv) < 2: print( "Usage: python3 -m scaffold ", file=sys.stderr, ) return 1 config_path = argv[1] try: with open(config_path, "r", encoding="utf-8") as fh: config = json.load(fh) except (OSError, json.JSONDecodeError) as exc: print(f"[scaffold] launch config error: {exc}", file=sys.stderr) return 1 # ---- Required §4a keys: missing => fail loud (no magic default) ---- try: path = config["path"] entrypoint = config["entrypoint"] interval = float(config["interval_seconds"]) default_signal_validity_seconds = int(config["default_signal_validity_seconds"]) wallet = config["wallet"] scanner_name = config["scanner_name"] except (KeyError, TypeError, ValueError) as exc: print( f"[scaffold] launch config missing/invalid required field: {exc}", file=sys.stderr, ) return 1 # ---- Optional §4a keys ---- timeout_seconds = config.get("timeout_seconds") # Hard cut: only `inputs` is read. The legacy `params` key is NOT honored. inputs = config.get("inputs") or {} state_path = config.get("state_path") state_history_max_count = config.get("state_history_max_count") or 0 # Per-signal data{} schema (§4a). None / absent => no scaffold-side validation # (intake still validates authoritatively). A dict turns on early, loud # scaffold-side validation that POSTs a violation to /errors. signal_data_schema = config.get("signal_data_schema") if signal_data_schema is not None and not isinstance(signal_data_schema, dict): signal_data_schema = None max_ticks = config.get("max_ticks") import scaffold as scaffold_pkg from scaffold import ScanEntrypointError, ScanSignatureError from scaffold.loop_primitives import log_event # Resolve run_scaffold via the package symbol so a test stub patched onto the # package (or this module) is picked up. run_scaffold = globals().get("run_scaffold") or scaffold_pkg.run_scaffold # Build the delivery sink when scanner_id and intake URL are available. sink = None scanner_id = os.environ.get("EXTERNAL_SCANNER_ID", "").strip() intake_url = os.environ.get("SIGNAL_INGESTION_URL", "").strip() if scanner_id and intake_url: try: from scaffold.delivery import build_delivery_sink post_timeout = float(os.environ.get("SIGNAL_POST_TIMEOUT_SECONDS", "5")) # Derive journal path from the state file directory when available. journal_path = None if state_path: journal_path = os.path.join( os.path.dirname(os.path.abspath(state_path)), "delivery-journal.jsonl", ) # §4a threads the per-data-key schema (`signal_data_schema`) to the # scaffold so the envelope validates each signal's `data` scaffold-side # and POSTs a violation to /errors (early, loud feedback). None ⇒ no # scaffold-side validation. The intake still performs authoritative # server-side schema validation regardless (this is additive). sink = build_delivery_sink( scanner_id=scanner_id, intake_url=intake_url, journal_path=journal_path, signal_data_schema=signal_data_schema, default_signal_validity_seconds=default_signal_validity_seconds, timeout=post_timeout, ) log_event("delivery_sink_enabled", scanner_id=scanner_id, intake_url=intake_url) except Exception as exc: # noqa: BLE001 log_event("delivery_sink_error", error=str(exc)) sink = None else: log_event( "delivery_sink_disabled", reason="EXTERNAL_SCANNER_ID or SIGNAL_INGESTION_URL not set; running with no-op sink", ) try: run_scaffold( path, entrypoint, interval=interval, timeout_seconds=timeout_seconds, inputs=inputs, state_path=state_path, scanner_name=scanner_name, wallet=wallet, default_signal_validity_seconds=default_signal_validity_seconds, state_history_max_count=state_history_max_count, sink=sink, max_ticks=max_ticks, ) except (ScanEntrypointError, ScanSignatureError, ValueError) as exc: # The full chain, not just the message. `import_entrypoint` raises `from exc`, so the real # cause — the missing sibling, the undefined name, the file and line it happened in — hangs # off __cause__. Printing only str(exc) discarded all of it and left an operator with # "failed to import scan.py" for a fault three modules deeper, in a file that was fine. print(f"[scaffold] launch failed: {exc}", file=sys.stderr) traceback.print_exception(type(exc), exc, exc.__traceback__, file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(_main())