diff --git a/chaski.py b/chaski.py index 3408693..5dce90c 100644 --- a/chaski.py +++ b/chaski.py @@ -342,7 +342,13 @@ def load_emitters(path: Path) -> list[tuple[emitter.Emitter, dict]]: if missing: raise RuleError(f"emitter {d.get('label', '?')!r} is missing {missing}") kinds = set(d["sink"]) - if kinds != {"jsonl"} and not ("graph" in kinds and kinds <= {"graph", "command", "alertmanager"}): + if kinds == {"scheduled"}: + from scheduled_sink_factory import validate + try: + validate(d["sink"]["scheduled"]) + except (ValueError, TypeError, OSError) as exc: + raise RuleError(f"emitter {d['label']}: invalid scheduled sink: {exc}") from None + elif kinds != {"jsonl"} and not ("graph" in kinds and kinds <= {"graph", "command", "alertmanager"}): raise RuleError(f"emitter {d['label']}: sink must be {{jsonl: }} or {{graph: {{...}}}}," " optionally with command: {...} and/or alertmanager: {...} beside graph") if "command" in kinds and not (d["sink"]["command"] or {}).get("argv"): @@ -394,6 +400,12 @@ def load_emitters(path: Path) -> list[tuple[emitter.Emitter, dict]]: raise RuleError("emitter trigger must be schedule, changes or external") if e.externally_triggered and d.get("baseline_emits") is not True: raise RuleError("external emitters require baseline_emits: true to retain their first event") + if kinds == {"scheduled"}: + from scheduled_sink_factory import validate_emitter + try: + validate_emitter(e) + except ValueError as exc: + raise RuleError(f"emitter {d['label']}: {exc}") from None out.append((e, d["sink"])) return out @@ -404,6 +416,9 @@ def make_sink(e: emitter.Emitter, sink: dict, quipu_url: str): never from the rules or the command line.""" import emitter as em + if set(sink) == {"scheduled"}: + from scheduled_sink_factory import build + return build(e, sink["scheduled"]) if "jsonl" in sink: return em.JsonlSink(sink["jsonl"]) import graph_sink @@ -481,18 +496,49 @@ def run_consumer(args) -> int: """Enter only after both persistent cursor and inbox custody are held.""" rules = load_rules(args.rules) reactor = Reactor(Quipu(args.quipu), rules, args.state, args.event_poll) - if args.emitters: - import emitter - - conn = emitter.connect(args.emitter_db or args.state.with_name("emitter.db")) - budget = emitter.WriteBudget(args.emitter_write_interval) # ONE, shared by all - from incremental import ChangeFeed, ChangeRunner - - reactor.runners = [(ChangeRunner if e.change_driven else emitter.Runner)( - e, conn, make_sink(e, sink, args.quipu), budget) for e, sink in load_emitters(args.emitters)] - changing = [r for r in reactor.runners if r.emitter.change_driven] - if changing: - reactor.change_feed = ChangeFeed(reactor.quipu, changing) + conn, sinks = None, [] + try: + if args.emitters: + import emitter + + conn = emitter.connect(args.emitter_db or args.state.with_name("emitter.db")) + budget = emitter.WriteBudget(args.emitter_write_interval) # ONE, shared by all + from incremental import ChangeFeed, ChangeRunner + + for e, definition in load_emitters(args.emitters): + sink = make_sink(e, definition, args.quipu) + sinks.append(sink) # retain custody even if runner construction fails + reactor.runners.append((ChangeRunner if e.change_driven else emitter.Runner)( + e, conn, sink, budget)) + changing = [r for r in reactor.runners if r.emitter.change_driven] + if changing: + reactor.change_feed = ChangeFeed(reactor.quipu, changing) + return _serve_consumer(args, reactor, rules) + finally: + # Drain before releasing ConsumerLease; pending outbox rows remain + # unacknowledged until their next delivery verifies the result. + try: + _close_sinks(sinks) + finally: + if conn is not None: + conn.close() + + +def _close_sinks(sinks): + error = None + for sink in reversed(sinks): + try: + close = getattr(sink, 'close', None) + if close is not None: + close() + except Exception as exc: + if error is None: + error = exc + if error is not None: + raise error + + +def _serve_consumer(args, reactor, rules): LOG.info("chaski: %d rule(s), event poll %ds", len(rules), reactor.event_poll_s) if args.once: now = time.time() diff --git a/docs/demos/scheduled-sink-factory.log b/docs/demos/scheduled-sink-factory.log new file mode 100644 index 0000000..0ab7a91 --- /dev/null +++ b/docs/demos/scheduled-sink-factory.log @@ -0,0 +1,11 @@ +Script started on 2026-10-15 09:18:55-04:00 [COMMAND="stty -onlcr && .venv/bin/python -I scripts/demo-scheduled-sink-factory.py" ] +Existing sink constructor positive control: True +Configured scheduled sink factory available: True +Submit delivery, acknowledged/pending: (0, 1) +After terminal publication, acknowledged/pending: (1, 0) +Phase receipts: ['admit', 'started', 'finished', 'publish'] +Explicit child outcome: {'outcome': 'skipped-by-hold', 'exit_code': 0} +Actual child invocations: 1 +No production graph, schedule, key or cron was used; signing is a fixture. + +Script done on 2026-10-15 09:18:55-04:00 [COMMAND_EXIT_CODE="0"] diff --git a/docs/demos/scheduled-sink-factory.timing b/docs/demos/scheduled-sink-factory.timing new file mode 100644 index 0000000..14a358a --- /dev/null +++ b/docs/demos/scheduled-sink-factory.timing @@ -0,0 +1,6 @@ +0.052777 49 +0.000014 50 +0.101955 46 +0.173618 57 +0.000054 131 +0.000013 103 diff --git a/docs/scheduled-sink-factory.md b/docs/scheduled-sink-factory.md new file mode 100644 index 0000000..ccab63b --- /dev/null +++ b/docs/scheduled-sink-factory.md @@ -0,0 +1,55 @@ +# Opt-in scheduled sink configuration + +An emitter may use the exclusive `sink: {scheduled: ...}` kind. It cannot be +combined with a graph, command, JSONL or alert sink. Shuttle's exact terminal +publication receipt is the acknowledgement boundary. Submission or a local +terminal result leaves the emitter's durable row pending. + +The emitter requires `trigger: changes`, `event: scheduled`, `baseline_emits: +true`, the default `item` key and BLOCKED-to-UNBLOCKED transition. Generation +rewriting, recovery, missing-as-from and renewal are refused: they would change +the canonical signed slot identity or acknowledge a different protocol. + +The scheduled mapping has exactly these fields: + +| Field | Meaning | +| --- | --- | +| `scope` | The six immutable `ExecutionScope` fields. | +| `ledger` | Absolute SQLite path in a provisioned, owned persistent directory. | +| `shuttle` | `executable`, `config`, `config_sha256`, `timeout_s` for the existing process port. | +| `execute` | Fixed `argv`, `cwd`, `env`, `result_fd_env`, `timeout_s`, `output_limit`. | +| `workers`, `capacity` | Explicit bounds, with `1 <= workers <= capacity <= 64`. | + +The signed scope's `manifest_sha256` must equal +`execution_manifest_sha256(execute)`: SHA256 of UTF-8 canonical JSON +`{"execute": RECIPE, "version": 1}` with sorted keys, compact separators and no +non-finite numbers. The literal recipe includes the complete explicit environment +and resource limits. Changing an argument, environment value or bound requires +a new digest and corresponding admitted declaration. The factory copies the +recipe so later mutation of the input mapping cannot change execution. + +This digest binds the recipe, not executable file bytes. Runtime admission must +still verify immutable artifact paths/content, parent directory custody and the +exact declaration/profile/query. The factory does not invent an eligibility +query, register keys or supply a catalogue adapter. Credentials belong in the +approved local execution policy and must never be printed as configuration. + +Before opening a ledger, the factory checks configuration shape, emitter policy, +recipe digest, ledger custody and the pinned Shuttle config's bytes/ownership. +The process port checks that config again through an opened descriptor on every +phase. Existing ledgers must be regular owned files, not symlinks. Directory +ancestry and shared-UID isolation remain operator responsibilities. + +Orderly consumer exit drains every constructed closeable sink before closing the +emitter database and releasing its consumer lease. Partial startup failures do +the same. A completed worker still needs outbox redelivery before acknowledgement. +Abrupt termination leaves claims unknown; cleanup is not permission to replay +execution. Preserve both the slot ledger and emitter database across rollback. +No claim of survival through supervisor shutdown or host reboot is made. + +`python scripts/demo-scheduled-sink-factory.py` uses actual temporary child +processes and SQLite outboxes, with an explicitly synthetic Shuttle executable. +The recorded result shows one hold outcome, one child invocation and four phase +receipts. This is source integration evidence, not live signing, profile loading, +schedule activation or deployment acceptance. The demo requires development test +dependencies because it reuses the isolated process fixture. diff --git a/pyproject.toml b/pyproject.toml index 9af3df7..0b18a16 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ chaski-once = "once:main" py-modules = [ "alertmanager_sink", "async_scheduled_sink", "change_stream", "chaski", "command_sink", "consumer_lease", "demo", "emitter", "external_event", "failure_family_adapter", - "fixed_outcome_executor", "graph_sink", "incremental", "once", "scheduled_event_sink", "scheduled_projection", "scheduled_receiver", "shuttle_process_port", "slot_receiver", + "fixed_outcome_executor", "graph_sink", "incremental", "once", "scheduled_event_sink", "scheduled_projection", "scheduled_receiver", "scheduled_sink_factory", "shuttle_process_port", "slot_receiver", ] [tool.pytest.ini_options] diff --git a/scheduled_sink_factory.py b/scheduled_sink_factory.py new file mode 100644 index 0000000..2fcb510 --- /dev/null +++ b/scheduled_sink_factory.py @@ -0,0 +1,102 @@ +"""Build an opt-in scheduled receiver from trusted local configuration only.""" +from __future__ import annotations + +from dataclasses import fields +import hashlib +import json +import os +from pathlib import Path +import re +import stat + +from async_scheduled_sink import AsyncScheduledSink +from fixed_outcome_executor import FixedOutcomeExecutor +from scheduled_event_sink import ScheduledEventSink +from scheduled_receiver import ExecutionScope +from shuttle_process_port import ShuttleProcessPort +from slot_receiver import SlotLedger + + +def _keys(value, expected): + if type(value) is not dict or set(value) != set(expected): + raise ValueError('complete supported scheduled sink configuration required') + + +def execution_manifest_sha256(recipe): + """Hash the literal versioned execution recipe; not executable file bytes.""" + raw = json.dumps({'version': 1, 'execute': recipe}, sort_keys=True, + separators=(',', ':'), allow_nan=False).encode() + return hashlib.sha256(raw).hexdigest() + + +def validate(config): + """Validate and copy fixed execution policy, without opening a ledger.""" + _keys(config, ('scope', 'ledger', 'shuttle', 'execute', 'workers', 'capacity')) + _keys(config['scope'], (f.name for f in fields(ExecutionScope))) + scope = ExecutionScope(**config['scope']) + scope.validate() + _keys(config['shuttle'], ('executable', 'config', 'config_sha256', 'timeout_s')) + shuttle = dict(config['shuttle']) + # Constructor validation has no subprocess, graph query or key operation. + ShuttleProcessPort(shuttle['executable'], shuttle['config'], shuttle['config_sha256'], + scope, 0, timeout_s=shuttle['timeout_s']) + _keys(config['execute'], ('argv', 'cwd', 'env', 'result_fd_env', 'timeout_s', 'output_limit')) + execute = FixedOutcomeExecutor(**config['execute']) + if execution_manifest_sha256(config['execute']) != scope.manifest_sha256: + raise ValueError('execution recipe differs from the signed manifest digest') + workers, capacity = config['workers'], config['capacity'] + if (type(workers) is not int or type(capacity) is not int or + not 1 <= workers <= capacity <= 64): + raise ValueError('require 1 <= workers <= capacity <= 64') + ledger = config['ledger'] + if type(ledger) is not str or not Path(ledger).is_absolute(): + raise ValueError('absolute persistent ledger path required') + return scope, ledger, shuttle, execute, workers, capacity + + +def _check_local_files(ledger, shuttle): + """Early feedback only; the process port rechecks pinned config per phase.""" + parent = Path(ledger).parent + info = parent.stat() + if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o022: + raise ValueError('provisioned owned ledger parent must not be writable by others') + if os.path.lexists(ledger): + info = Path(ledger).lstat() + if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o022: + raise ValueError('existing ledger must be an owned regular file') + fd = os.open(shuttle['config'], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + with os.fdopen(fd, 'rb') as source: + info = os.fstat(source.fileno()) + if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or + info.st_mode & 0o022 or info.st_size > 65536): + raise ValueError('pinned Shuttle config custody unavailable') + raw = source.read(65537) + if len(raw) > 65536 or hashlib.sha256(raw).hexdigest() != shuttle['config_sha256']: + raise ValueError('Shuttle config differs from pinned bytes') + + +def validate_emitter(emitter): + if (emitter.event != 'scheduled' or not emitter.baseline_emits or + emitter.generation_event_ids or emitter.resolve_recovery or + emitter.missing_is_from or emitter.renew_interval_s or not emitter.change_driven or + emitter.externally_triggered or emitter.key != 'item' or + emitter.from_verdict != 'BLOCKED' or emitter.to_verdict != 'UNBLOCKED' or + type(emitter.label) is not str or + not re.fullmatch(r'[A-Za-z][A-Za-z0-9_.-]{0,127}', emitter.label) or + type(emitter.owner) is not str or not 0 < len(emitter.owner) <= 256): + raise ValueError('scheduled sink requires canonical non-recovering scheduled events') + + +def build(emitter, config): + scope, ledger, shuttle, execute, workers, capacity = validate(config) + validate_emitter(emitter) + _check_local_files(ledger, shuttle) + receiver = AsyncScheduledSink(SlotLedger(ledger), scope, + lambda slot: ShuttleProcessPort(shuttle['executable'], shuttle['config'], + shuttle['config_sha256'], scope, slot, timeout_s=shuttle['timeout_s']), + execute, workers=workers, capacity=capacity) + try: + return ScheduledEventSink(receiver, emitter.label, emitter.owner) + except BaseException: + receiver.close() + raise diff --git a/scripts/demo-scheduled-sink-factory.py b/scripts/demo-scheduled-sink-factory.py new file mode 100644 index 0000000..3cc4be3 --- /dev/null +++ b/scripts/demo-scheduled-sink-factory.py @@ -0,0 +1,48 @@ +"""Actual local processes and outbox; Shuttle admission/signing is a fixture.""" +import importlib.util +import json +import logging +from pathlib import Path +import sys +import tempfile + +source = (Path(sys.argv[2]).resolve() if len(sys.argv) == 3 and sys.argv[1] == '--source' + else Path(__file__).resolve().parents[1]) +sys.path.insert(0, str(source)) +from chaski import make_sink # noqa: E402 + +print('Existing sink constructor positive control:', callable(make_sink)) +present = importlib.util.find_spec('scheduled_sink_factory') is not None +print('Configured scheduled sink factory available:', present) +if not present: + sys.exit(0) +from emitter import connect, deliver_pending, observe # noqa: E402 +from scheduled_receiver import run_id # noqa: E402 +from test_scheduled_sink_factory import configured # noqa: E402 + +# The acknowledgement counts below show the expected asynchronous pending +# state; omit its long canonical event identifier from this short recording. +logging.getLogger('chaski.emitter').setLevel(logging.ERROR) +with tempfile.TemporaryDirectory() as root: + load, _, _, scope, trace, counter, _, _ = configured.__wrapped__(Path(root)) + definition, options = load() + sink = make_sink(definition, options, 'http://unused.invalid') + conn = connect(Path(root)/'outbox.sqlite') + try: + observe(conn, definition, [dict(item=scope.job, owner=definition.owner, + verdict='UNBLOCKED', event_id=run_id(scope, 600))], 600) + print('Submit delivery, acknowledged/pending:', + deliver_pending(conn, sink, 600, definition.label)) + with sink.receiver._lock: + future = next(iter(sink.receiver._pending.values())) + future.result(timeout=5) + print('After terminal publication, acknowledged/pending:', + deliver_pending(conn, sink, 1600, definition.label)) + phases = [json.loads(x) for x in trace.read_text().splitlines()] + print('Phase receipts:', [p['action'] for p in phases]) + print('Explicit child outcome:', phases[2]['packet']['outcome']) + print('Actual child invocations:', counter.read_text()) + print('No production graph, schedule, key or cron was used; signing is a fixture.') + finally: + sink.close() + conn.close() diff --git a/test_scheduled_sink_factory.py b/test_scheduled_sink_factory.py new file mode 100644 index 0000000..1b2b8ba --- /dev/null +++ b/test_scheduled_sink_factory.py @@ -0,0 +1,346 @@ +"""Opt-in factory with actual subprocesses/outbox; Shuttle CLI is a test double.""" + +from dataclasses import asdict, replace +from pathlib import Path +import hashlib +import json +import sys + +import pytest +import yaml + +import chaski +import incremental +from emitter import Emitter, connect, deliver_pending, observe +from scheduled_receiver import run_id +from scheduled_sink_factory import execution_manifest_sha256 +from shuttle_process_port import PortRefused +from test_shuttle_process_port import setup as port_fixture + + +@pytest.fixture +def configured(tmp_path): + _, scope, trace, config, executable = port_fixture.__wrapped__(tmp_path) + # A real child with an owned counter, never the production program. + counter = tmp_path / "child-calls" + program = ( + "import os,json;from pathlib import Path;" + f'p=Path({str(counter)!r});p.write_text(str(int(p.read_text())+1) if p.exists() else "1");' + 'os.write(int(os.environ["RESULT_FD"]),json.dumps(dict(' + 'schema_version=1,outcome="skipped-by-hold",exit_code=0)).encode())' + ) + spec = dict( + scope=asdict(scope), + ledger=str(tmp_path / "slots.sqlite"), + shuttle=dict( + executable=str(executable), + config=str(config), + config_sha256=hashlib.sha256(config.read_bytes()).hexdigest(), + timeout_s=5, + ), + execute=dict( + argv=[sys.executable, "-c", program], + cwd=str(tmp_path), + env={}, + result_fd_env="RESULT_FD", + timeout_s=2, + output_limit=1048576, + ), + workers=1, + capacity=2, + ) + scope = replace(scope, manifest_sha256=execution_manifest_sha256(spec["execute"])) + spec["scope"] = asdict(scope) + definition = dict( + label="fixture", + command=["unused"], + schedule="PT1H", + owner="operator", + trigger="changes", + event="scheduled", + baseline_emits=True, + sink={"scheduled": spec}, + ) + path = tmp_path / "emitters.yml" + + def load(): + path.write_text(yaml.safe_dump({"emitters": [definition]})) + return chaski.load_emitters(path)[0] + + return load, definition, spec, scope, trace, counter, config, executable + + +def test_actual_outbox_factory_waits_for_terminal_process_receipt(configured, tmp_path): + load, _, spec, scope, trace, counter, _, _ = configured + e, d = load() + sink = chaski.make_sink(e, d, "http://unused.invalid") + spec["execute"]["argv"][:] = [sys.executable, "-c", "raise SystemExit(9)"] + conn = connect(tmp_path / "outbox.sqlite") + try: + observe(conn, e, [dict(item=scope.job, owner=e.owner, verdict="UNBLOCKED", event_id=run_id(scope, 600))], 600) + assert deliver_pending(conn, sink, 600, e.label) == (0, 1) + assert conn.execute("SELECT delivered_at FROM outbox").fetchone() == (None,) + with sink.receiver._lock: + future = next(iter(sink.receiver._pending.values())) + future.result(timeout=5) + assert deliver_pending(conn, sink, 1600, e.label) == (1, 0) + assert counter.read_text() == "1" + events = [json.loads(x) for x in trace.read_text().splitlines()] + assert [x["action"] for x in events] == ["admit", "started", "finished", "publish"] + assert events[2]["packet"]["outcome"] == {"outcome": "skipped-by-hold", "exit_code": 0} + assert deliver_pending(conn, sink, 2600, e.label) == (0, 0) + finally: + sink.close() + conn.close() + assert spec["execute"]["env"] == {} # no descriptor or ambient environment leaked into policy + + +def test_lost_publication_receipt_retries_publication_without_child_replay(configured, tmp_path): + load, _, spec, scope, trace, counter, config, executable = configured + config.write_text(json.dumps({"trace": str(trace), "mode": "badpublish"})) + spec["shuttle"]["config_sha256"] = hashlib.sha256(config.read_bytes()).hexdigest() + executable.write_text( + executable.read_text().replace( + "if mode == 'fail': sys.exit(3)", + "if mode == 'fail' or (mode == 'badpublish' and action == 'publish' and " + "sum(json.loads(x)['action']=='publish' for x in " + "Path(config['trace']).read_text().splitlines())==1): sys.exit(3)", + ) + ) + e, d = load() + sink = chaski.make_sink(e, d, "http://unused.invalid") + conn = connect(tmp_path / "outbox.sqlite") + try: + observe(conn, e, [dict(item=scope.job, owner=e.owner, verdict="UNBLOCKED", event_id=run_id(scope, 600))], 600) + assert deliver_pending(conn, sink, 600, e.label) == (0, 1) + with sink.receiver._lock: + future = next(iter(sink.receiver._pending.values())) + with pytest.raises(PortRefused): + future.result(timeout=5) + assert deliver_pending(conn, sink, 1600, e.label) == (0, 1) # consume actual worker error + assert deliver_pending(conn, sink, 2600, e.label) == (0, 1) # publication replay submitted + with sink.receiver._lock: + future = next(iter(sink.receiver._pending.values())) + future.result(timeout=5) + assert deliver_pending(conn, sink, 3600, e.label) == (1, 0) + assert counter.read_text() == "1" + events = [json.loads(x)["action"] for x in trace.read_text().splitlines()] + assert events == ["admit", "started", "finished", "publish", "admit", "publish"] + finally: + sink.close() + conn.close() + + +@pytest.mark.parametrize("change", ["changed", "symlink", "fifo", "writable", "ledger-symlink"]) +def test_local_custody_refuses_before_ledger_or_any_process(configured, tmp_path, change): + import os + + load, _, spec, _, trace, counter, config, _ = configured + e, d = load() + if change == "changed": + config.write_text("{}") + elif change == "writable": + config.chmod(0o666) + elif change == "ledger-symlink": + Path(spec["ledger"]).symlink_to(tmp_path / "not-a-ledger") + else: + config.unlink() + if change == "symlink": + config.symlink_to(tmp_path / "elsewhere") + else: + os.mkfifo(config) + with pytest.raises((ValueError, OSError)): + chaski.make_sink(e, d, "http://unused.invalid") + assert not trace.exists() and not counter.exists() + assert not (tmp_path / "not-a-ledger").exists() + if change != "ledger-symlink": + assert not (tmp_path / "slots.sqlite").exists() + + +@pytest.mark.parametrize( + "field,value", + [ + ("workers", True), + ("workers", 0), + ("capacity", 65), + ("ledger", "relative"), + ("scope", {}), + ("execute", {}), + ("shuttle", {}), + ("extra", 1), + ], +) +def test_bad_configuration_refuses_during_load_without_creating_state(configured, tmp_path, field, value): + load, _, spec, _, trace, counter, _, _ = configured + spec[field] = value + with pytest.raises(chaski.RuleError): + load() + assert not (tmp_path / "slots.sqlite").exists() and not trace.exists() and not counter.exists() + + +@pytest.mark.parametrize( + "field,value", + [ + ("event", "unblocked"), + ("baseline_emits", False), + ("generation_event_ids", True), + ("trigger", "schedule"), + ("label", "invalid label"), + ("key", "foreign"), + ("from", "OK"), + ("to", "BAD"), + ("resolve_recovery", True), + ], +) +def test_incompatible_emitter_policy_refuses(configured, field, value): + load, definition, *_ = configured + definition[field] = value + with pytest.raises(chaski.RuleError): + load() + + +def test_consumer_closes_every_sink_and_database_before_releasing_lease(tmp_path, monkeypatch): + import emitter + from consumer_lease import ConsumerLease, ConsumerLeaseError + + state, inbox = tmp_path / "cursor.json", tmp_path / "emitter.sqlite" + events = [] + + class Sink: + def __init__(self, name): + self.name = name + + def close(self): + with pytest.raises(ConsumerLeaseError): + with ConsumerLease(state, inbox): + pass + events.append(self.name) + + class Conn: + def close(self): + events.append("database") + + monkeypatch.setattr(chaski, "load_rules", lambda _: []) + monkeypatch.setattr( + chaski, + "load_emitters", + lambda _: [(Emitter("one", ["unused"], 60, "operator"), {}), (Emitter("two", ["unused"], 60, "operator"), {})], + ) + monkeypatch.setattr(chaski, "make_sink", lambda e, *_: Sink(e.label)) + monkeypatch.setattr(emitter, "connect", lambda _: Conn()) + + class Runner: + def __init__(self, e, *_): + self.emitter = e + + monkeypatch.setattr(emitter, "Runner", Runner) + monkeypatch.setattr(chaski, "_serve_consumer", lambda *_: events.append("serve") or 0) + assert ( + chaski.main( + [ + "--quipu", + "http://unused.invalid", + "--rules", + str(tmp_path / "rules.yml"), + "--state", + str(state), + "--emitters", + str(tmp_path / "emitters.yml"), + "--emitter-db", + str(inbox), + "--once", + ] + ) + == 0 + ) + assert events == ["serve", "two", "one", "database"] + with ConsumerLease(state, inbox): + pass + + +def test_shutdown_failure_still_drains_other_owned_sinks(): + calls = [] + + class Sink: + def __init__(self, name): + self.name = name + + def close(self): + calls.append(self.name) + if self.name == "two": + raise RuntimeError("synthetic close failure") + + with pytest.raises(RuntimeError, match="synthetic"): + chaski._close_sinks([Sink("one"), Sink("two")]) + assert calls == ["two", "one"] + + +@pytest.mark.parametrize( + "field,value", + [ + ("argv", [sys.executable, "-c", "pass"]), + ("env", {"EXAMPLE": "changed"}), + ("timeout_s", 3), + ("output_limit", 1024), + ], +) +def test_changed_literal_recipe_cannot_keep_old_manifest_digest(configured, tmp_path, field, value): + load, _, spec, _, trace, counter, _, _ = configured + spec["execute"][field] = value + with pytest.raises(chaski.RuleError, match="manifest digest"): + load() + assert not trace.exists() and not counter.exists() and not (tmp_path / "slots.sqlite").exists() + + +def test_partial_startup_failure_closes_constructed_sinks_and_database(tmp_path, monkeypatch): + import emitter + + events = [] + + class Conn: + def close(self): + events.append("database") + + class Sink: + def __init__(self, name): + self.name = name + + def close(self): + events.append(self.name) + + class Runner: + def __init__(self, e, *_): + self.emitter = e + + def failed(*_): + raise RuntimeError("synthetic adapter startup failure") + + monkeypatch.setattr(chaski, "load_rules", lambda _: []) + monkeypatch.setattr( + chaski, + "load_emitters", + lambda _: [ + (Emitter("one", ["unused"], 60, "operator"), {}), + (Emitter("two", ["unused"], 60, "operator", change_driven=True), {}), + ], + ) + monkeypatch.setattr(chaski, "make_sink", lambda e, *_: Sink(e.label)) + monkeypatch.setattr(emitter, "connect", lambda _: Conn()) + monkeypatch.setattr(emitter, "Runner", Runner) + monkeypatch.setattr(incremental, "ChangeRunner", failed) + args = type( + "Args", + (), + dict( + quipu="http://unused.invalid", + rules=tmp_path / "rules.yml", + state=tmp_path / "cursor.json", + event_poll=60, + emitters=tmp_path / "emitters.yml", + emitter_db=tmp_path / "emitter.sqlite", + emitter_write_interval=0, + ), + )() + with pytest.raises(RuntimeError, match="adapter startup"): + chaski.run_consumer(args) + assert events == ["two", "one", "database"]