Skills › Compound Engineering › ce-sweep › sweep-state.py
sweep-state.py
793 lines · skills/ce-sweep/scripts/sweep-state.py at 3449230
1#!/usr/bin/env python3
2"""Deterministic state engine for the feedback sweep (ce-sweep).
4This helper owns ALL reads and writes of the sweep state file. Peer agents
5(source connectors, the analyzer, the orchestrator) never edit the file by
6hand — they go through these subcommands so the schema contract, the id-keyed
7merge, the single-writer lease, and the closed-item evidence rule are enforced
8in exactly one place. See `references/state-schema.md` for the cross-agent
9contract this script implements.
11Design rules (shared with the repo's other state helpers):
12 - Pure Python 3 stdlib. No third-party dependencies.
13 - Every OPERATIONAL failure path prints a parseable STATUS WORD on line 1 and
14 exits 0 — it never raises a traceback to the caller. Only genuine CLI
15 misuse (bad/missing subcommand args) exits non-zero via argparse.
16 - Writes are atomic: a temp file in the state dir + os.replace (atomic on
17 POSIX), so a concurrent reader never sees a torn file.
18 - The script never calls the wall clock for the values it stores EXCEPT the
19 lease timestamp (staleness needs "now"). Tests pin it with --now / stamp
20 values with --timestamp so behavior is reproducible.
22STATUS WORDS (line 1 of stdout for every subcommand):
23 OK success (an optional JSON payload follows on line 2)
24 NO-STATE read: the state file does not exist yet
25 CORRUPT the state file exists but does not parse as our schema
26 LOCKED lease-acquire: a live lease is held by another writer
27 STALE-RECLAIMED lease-acquire: an expired lease was taken over (JSON payload)
28 LEASE-LOST a mutating call was made by a writer that does not own the
29 lease (or a release of another writer's lease); no write
30 REFUSED cursor-advance: unknown past-item, or non-monotonic cursor
31 ERROR an unexpected internal error (defensive; never a traceback)
33The state file is genuine YAML restricted to a small, deliberate subset so a
34hand-written stdlib serializer/parser round-trips it deterministically. See
35`references/state-schema.md` (section "YAML subset") for the exact grammar:
36non-empty dict values become block mappings (any depth); scalars are emitted as
37JSON tokens (strings always double-quoted on one line); lists and empty dicts
38are emitted as inline JSON flow on a single line — itself valid YAML.
39"""
40import argparse
41import json
42import os
43import sys
44import tempfile
45from datetime import datetime, timezone
47try:
48 import fcntl # POSIX advisory locks (macOS, Linux — this repo's Unix targets)
49 _HAS_FCNTL = True
50except ImportError: # non-POSIX; degrade to unlocked (single-writer by convention)
51 _HAS_FCNTL = False
53SCHEMA_VERSION = 1
55# The closed lifecycle status enum is documented in references/state-schema.md.
56# Unknown statuses are preserved on write-back, never dropped — nothing here
57# whitelists values.
59# A `closed` item MUST carry all three evidence fields; `validate` downgrades
60# any closed item missing any of them back to `fix_pending`.
61EVIDENCE_FIELDS = ("fix_ref", "verified_merge_sha", "verified_at")
63DEFAULT_TTL_MINUTES = 60
66# --------------------------------------------------------------------------- #
67# Minimal YAML subset: serializer + parser
68# --------------------------------------------------------------------------- #
69# Grammar (2-space indent, no tabs):
70# - Block mappings: `<key>: <value>` or `<key>:` (nested mapping follows).
71# - Keys: a bare token matching _SAFE_KEY, else a JSON double-quoted string.
72# - Scalars: null / true / false / integers / floats emitted bare; strings
73# always emitted as JSON double-quoted (single line, fully escaped).
74# - Nested containers (dict/list) that are NOT part of the known block
75# structure are emitted as inline JSON flow on one line — valid YAML.
76# The emitter only produces block MAPPINGS (never block sequences); the parser
77# tolerates arbitrary-depth block mappings plus scalar / inline-JSON values.
79import re
81_SAFE_KEY = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*$")
83# Preferred key orders for deterministic, human-legible output. Any keys not
84# listed are appended in sorted order so unknown/preserved fields stay stable.
85_DOC_ORDER = ("schema_version", "lease", "sources", "items", "last_run")
86_LEASE_ORDER = ("writer", "timestamp", "ttl_minutes")
87_ITEM_ORDER = (
88 "source", "status", "sensitive", "title", "url", "body", "quote",
89 "fix_ref", "verified_merge_sha", "verified_at",
90)
91_LAST_RUN_ORDER = ("timestamp", "outcome", "writer", "counts")
94def _ordered_keys(d, preferred):
95 seen = [k for k in preferred if k in d]
96 rest = sorted(k for k in d if k not in preferred)
97 return seen + rest
100def _emit_key(key):
101 key = str(key)
102 if _SAFE_KEY.match(key):
103 return key
104 return json.dumps(key, ensure_ascii=False)
107def _emit_scalar(v):
108 if v is None:
109 return "null"
110 if v is True:
111 return "true"
112 if v is False:
113 return "false"
114 if isinstance(v, int):
115 return str(v)
116 if isinstance(v, float):
117 return repr(v)
118 if isinstance(v, str):
119 return json.dumps(v, ensure_ascii=False)
120 # dict / list -> inline JSON flow (valid YAML), sorted for determinism.
121 return json.dumps(v, ensure_ascii=False, sort_keys=True)
124def _emit_mapping(d, indent, preferred):
125 """Emit a dict as block YAML. Only nests block-style for child dicts;
126 lists and any other container leaf are emitted inline via _emit_scalar."""
127 pad = " " * indent
128 lines = []
129 for key in _ordered_keys(d, preferred):
130 val = d[key]
131 ks = _emit_key(key)
132 if isinstance(val, dict) and val:
133 lines.append(f"{pad}{ks}:")
134 # Nested item/source/lease/last_run maps: no special preferred order
135 # for arbitrary nested dicts; only the top structural maps below get
136 # a preferred order (handled by their callers).
137 child_pref = _child_preferred(indent, key)
138 lines.extend(_emit_mapping(val, indent + 1, child_pref))
139 else:
140 lines.append(f"{pad}{ks}: {_emit_scalar(val)}")
141 return lines
144def _child_preferred(parent_indent, key):
145 """Preferred child-key order for the known structural containers."""
146 if parent_indent == 0:
147 if key == "lease":
148 return _LEASE_ORDER
149 if key == "last_run":
150 return _LAST_RUN_ORDER
151 return () # sources / items: children are id-keyed, no field order
152 # A source entry or item entry: apply the item field order (harmless for
153 # source entries, which only carry `cursor`/`sensitive`).
154 return _ITEM_ORDER
157def emit_document(state):
158 lines = _emit_mapping(state, 0, _DOC_ORDER)
159 return "\n".join(lines) + "\n"
162def _split_key(content):
163 """Split a mapping line into (key, rest-after-colon). Raises ValueError on
164 a line with no colon so a malformed file surfaces as CORRUPT."""
165 if content.startswith('"'):
166 key, end = json.JSONDecoder().raw_decode(content)
167 rest = content[end:]
168 if not rest.startswith(":"):
169 raise ValueError("expected ':' after quoted key")
170 return key, rest[1:]
171 idx = content.find(":")
172 if idx == -1:
173 raise ValueError("expected ':' in mapping line")
174 return content[:idx], content[idx + 1:]
177_BLOCK = object() # sentinel: value is a nested block, parse deeper lines
180def _parse_value(rest):
181 rest = rest.strip()
182 if rest == "":
183 return _BLOCK
184 if rest == "null":
185 return None
186 if rest == "true":
187 return True
188 if rest == "false":
189 return False
190 first = rest[0]
191 if first in '"{[':
192 return json.loads(rest) # JSON string, object, or array (flow style)
193 if first == "-" or first.isdigit():
194 try:
195 if "." in rest or "e" in rest or "E" in rest:
196 return float(rest)
197 return int(rest)
198 except ValueError:
199 pass
200 # Emitter never produces bare unquoted strings; a bare token here is a
201 # hand-edit — keep it verbatim rather than failing.
202 return rest
205def _parse_mapping(rows, cursor, indent):
206 result = {}
207 while cursor["i"] < len(rows):
208 cur_indent, content = rows[cursor["i"]]
209 if cur_indent < indent:
210 break
211 if cur_indent > indent:
212 # Orphan deeper line with no parent key; skip defensively.
213 cursor["i"] += 1
214 continue
215 cursor["i"] += 1
216 key, rest = _split_key(content)
217 val = _parse_value(rest)
218 if val is _BLOCK:
219 if cursor["i"] < len(rows) and rows[cursor["i"]][0] > indent:
220 val = _parse_mapping(rows, cursor, rows[cursor["i"]][0])
221 else:
222 val = {}
223 result[key] = val
224 return result
227def parse_document(text):
228 rows = []
229 for raw in text.split("\n"):
230 stripped = raw.strip()
231 if not stripped or stripped.startswith("#"):
232 continue
233 indent = len(raw) - len(raw.lstrip(" "))
234 rows.append((indent, stripped))
235 if not rows:
236 return {}
237 return _parse_mapping(rows, {"i": 0}, rows[0][0])
240# --------------------------------------------------------------------------- #
241# State load / save
242# --------------------------------------------------------------------------- #
244def _item_key(source, item_id):
245 """Storage key for an item. The items keyspace is flat, so a bare id can
246 collide across sources (two Slack channels emitting the same ts, or a
247 source that reuses numeric ids). Namespacing by source keeps each source's
248 id space independent, matching the composite keys documented in
249 references/state-schema.md."""
250 return "{}:{}".format(source, item_id)
253def load_state(path):
254 """Return (status, data): ('absent', None), ('corrupt', None), or
255 ('ok', dict). A file that parses but lacks schema_version is corrupt."""
256 try:
257 with open(path, encoding="utf-8") as f:
258 # A machine-local state file can live under world-shared /tmp, and
259 # it is a correctness dependency (lease, cursors, closed status) as
260 # well as an injection sink (item bodies re-read into agent
261 # context). Reject a file not owned by us so a co-tenant cannot
262 # plant a forged lease/cursor or attacker-authored item text. Skip
263 # where geteuid is unavailable (non-POSIX), where the threat does
264 # not apply.
265 geteuid = getattr(os, "geteuid", None)
266 if geteuid is not None and os.fstat(f.fileno()).st_uid != geteuid():
267 return ("corrupt", None)
268 text = f.read()
269 except FileNotFoundError:
270 return ("absent", None)
271 except (OSError, UnicodeDecodeError):
272 return ("corrupt", None)
273 if not text.strip():
274 return ("absent", None)
275 try:
276 data = parse_document(text)
277 except Exception:
278 return ("corrupt", None)
279 if not isinstance(data, dict) or "schema_version" not in data:
280 return ("corrupt", None)
281 data.setdefault("sources", {})
282 data.setdefault("items", {})
283 return ("ok", data)
286def new_state():
287 return {"schema_version": SCHEMA_VERSION, "sources": {}, "items": {}}
290def write_state(path, state):
291 """Atomic write of the state file. Returns True on success."""
292 state["schema_version"] = SCHEMA_VERSION
293 text = emit_document(state)
294 d = os.path.dirname(os.path.abspath(path))
295 os.makedirs(d, exist_ok=True)
296 fd, tmp = tempfile.mkstemp(dir=d, prefix=".tmp-sweep-", suffix=".yml")
297 try:
298 with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as f:
299 f.write(text)
300 os.replace(tmp, path)
301 except BaseException:
302 try:
303 os.unlink(tmp)
304 except OSError:
305 pass
306 raise
307 return True
310# --------------------------------------------------------------------------- #
311# Output + time helpers
312# --------------------------------------------------------------------------- #
314def emit(status_word, payload=None):
315 print(status_word)
316 if payload is not None:
317 print(json.dumps(payload))
318 return 0
321def resolve_now(args):
322 """The 'current time' for lease staleness + lease re-stamping. Pinned by
323 --now in tests; otherwise the real UTC clock."""
324 now = getattr(args, "now", None)
325 if now:
326 return now
327 return datetime.now(timezone.utc).isoformat()
330def _parse_iso(s):
331 if not isinstance(s, str) or not s:
332 return None
333 try:
334 dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
335 except ValueError:
336 return None
337 if dt.tzinfo is None:
338 dt = dt.replace(tzinfo=timezone.utc)
339 return dt
342def current_lease(state):
343 """The active lease dict (with a non-empty writer), or None."""
344 lease = state.get("lease")
345 if isinstance(lease, dict) and lease.get("writer"):
346 return lease
347 return None
350def lease_is_stale(lease, now_iso):
351 """True only when we can PROVE the lease is older than its TTL. If either
352 timestamp is unparseable we cannot prove staleness -> treat as live (do not
353 reclaim). Conservative: never stomp a lease we cannot show is expired."""
354 ts = _parse_iso(lease.get("timestamp", ""))
355 now = _parse_iso(now_iso)
356 if ts is None or now is None:
357 return False
358 try:
359 ttl = int(lease.get("ttl_minutes", DEFAULT_TTL_MINUTES))
360 except (TypeError, ValueError):
361 ttl = DEFAULT_TTL_MINUTES
362 return (now - ts).total_seconds() > ttl * 60
365def restamp_lease(state, writer, now_iso):
366 """Refresh the owning writer's lease timestamp so a long sweep keeps it
367 alive across many writes. Only called after ownership is confirmed."""
368 lease = state.get("lease")
369 if isinstance(lease, dict) and lease.get("writer") == writer:
370 lease["timestamp"] = now_iso
373def owns_lease(state, writer):
374 lease = current_lease(state)
375 return lease is not None and lease.get("writer") == writer
378# --------------------------------------------------------------------------- #
379# Subcommands
380# --------------------------------------------------------------------------- #
382def cmd_read(args):
383 st, data = load_state(args.state)
384 if st == "absent":
385 return emit("NO-STATE")
386 if st == "corrupt":
387 return emit("CORRUPT")
388 return emit("OK", data)
391def cmd_validate(args):
392 st, data = load_state(args.state)
393 if st == "corrupt":
394 return emit("CORRUPT")
395 if st == "absent":
396 return emit("OK", {"downgraded": []})
397 downgraded = []
398 for item_id, item in data.get("items", {}).items():
399 if not isinstance(item, dict):
400 continue
401 if item.get("status") == "closed" and any(
402 not item.get(f) for f in EVIDENCE_FIELDS
403 ):
404 item["status"] = "fix_pending"
405 downgraded.append(item_id)
406 if downgraded:
407 write_state(args.state, data)
408 return emit("OK", {"downgraded": sorted(downgraded)})
411def _load_owned_state(args):
412 """Load state for a lease-gated mutation. Returns (data, None) when
413 args.writer holds the lease, else (None, status_word) for the caller to
414 emit. Absent state means no lease to own -> the caller is not the owner."""
415 st, data = load_state(args.state)
416 if st == "corrupt":
417 return None, "CORRUPT"
418 if st == "absent" or not owns_lease(data, args.writer):
419 return None, "LEASE-LOST"
420 return data, None
423def _commit_owned(args, data):
424 """Shared tail for lease-gated mutations: re-stamp the lease, persist."""
425 restamp_lease(data, args.writer, resolve_now(args))
426 write_state(args.state, data)
427 return emit("OK")
430def cmd_upsert_item(args):
431 data, err = _load_owned_state(args)
432 if err:
433 return emit(err)
434 try:
435 incoming = json.loads(args.json)
436 except (ValueError, TypeError):
437 return emit("ERROR")
438 if not isinstance(incoming, dict):
439 return emit("ERROR")
441 items = data.setdefault("items", {})
442 key = _item_key(args.source, args.id)
443 existing = items.get(key)
444 merged = dict(existing) if isinstance(existing, dict) else {}
445 # id-keyed merge: only the keys present in the incoming item are replaced;
446 # unknown fields already on the item survive untouched.
447 merged.update(incoming)
448 merged["source"] = args.source
449 merged["id"] = args.id
451 source_entry = data.get("sources", {}).get(args.source, {})
452 is_sensitive = (
453 merged.get("sensitive") is True
454 or (isinstance(source_entry, dict) and source_entry.get("sensitive") is True)
455 )
456 if is_sensitive:
457 for f in ("body", "quote"):
458 merged.pop(f, None)
460 items[key] = merged
461 return _commit_owned(args, data)
464def cmd_cursor_get(args):
465 st, data = load_state(args.state)
466 if st == "corrupt":
467 return emit("CORRUPT")
468 if st == "absent":
469 return emit("OK", {"cursor": None})
470 entry = data.get("sources", {}).get(args.source, {})
471 cursor = entry.get("cursor") if isinstance(entry, dict) else None
472 return emit("OK", {"cursor": cursor})
475def cmd_cursor_advance(args):
476 data, err = _load_owned_state(args)
477 if err:
478 return emit(err)
479 # The cursor may only advance past an item that actually exists in state,
480 # so a resume never skips unrecorded items.
481 if _item_key(args.source, args.past_item) not in data.get("items", {}):
482 return emit("REFUSED")
483 entry = data.setdefault("sources", {}).setdefault(args.source, {})
484 current = entry.get("cursor")
485 # Monotonic guard. A new cursor sorting strictly before the current one is
486 # refused; equal is allowed (idempotent re-advance).
487 if current is not None and _cursor_lt(str(args.to), str(current)):
488 return emit("REFUSED")
489 entry["cursor"] = args.to
490 return _commit_owned(args, data)
493def _cursor_lt(a, b):
494 """True when cursor `a` precedes `b`. Pure-digit cursors compare
495 numerically so an unpadded id like '9' correctly precedes '10'; everything
496 else (Slack fixed-width ts, ISO timestamps) already sorts correctly as a
497 string, so fall back to lexical order."""
498 if a.isdigit() and b.isdigit():
499 return int(a) < int(b)
500 return a < b
503def cmd_lease_acquire(args):
504 st, data = load_state(args.state)
505 if st == "corrupt":
506 return emit("CORRUPT")
507 if st == "absent":
508 data = new_state()
509 now = resolve_now(args)
510 ttl = args.ttl_minutes if args.ttl_minutes is not None else DEFAULT_TTL_MINUTES
511 lease = current_lease(data)
512 if lease is None or lease.get("writer") == args.writer:
513 # Free, or re-entrant acquire by the same writer: (re)stamp and take it.
514 data["lease"] = {"writer": args.writer, "timestamp": now, "ttl_minutes": ttl}
515 write_state(args.state, data)
516 return emit("OK")
517 if lease_is_stale(lease, now):
518 prev = {
519 "previous_writer": lease.get("writer"),
520 "previous_timestamp": lease.get("timestamp"),
521 }
522 data["lease"] = {"writer": args.writer, "timestamp": now, "ttl_minutes": ttl}
523 write_state(args.state, data)
524 return emit("STALE-RECLAIMED", prev)
525 return emit("LOCKED")
528def cmd_lease_release(args):
529 st, data = load_state(args.state)
530 if st == "corrupt":
531 return emit("CORRUPT")
532 if st == "absent":
533 return emit("OK") # nothing to release
534 lease = current_lease(data)
535 if lease is None:
536 return emit("OK")
537 if lease.get("writer") != args.writer:
538 return emit("LEASE-LOST") # never release another writer's lease
539 data.pop("lease", None)
540 write_state(args.state, data)
541 return emit("OK")
544def cmd_run_record(args):
545 # Intentionally lease-agnostic: an `aborted-locked` run could not acquire
546 # the lease yet must still record its outcome. In local-commit mode there
547 # is a single writer per checkout, so this bookkeeping write is safe.
548 st, data = load_state(args.state)
549 if st == "corrupt":
550 return emit("CORRUPT")
551 if st == "absent":
552 data = new_state()
553 try:
554 counts = json.loads(args.counts)
555 except (ValueError, TypeError):
556 counts = {}
557 data["last_run"] = {
558 "timestamp": args.timestamp,
559 "outcome": args.outcome,
560 "writer": args.writer,
561 "counts": counts,
562 }
563 write_state(args.state, data)
564 return emit("OK")
567def cmd_import_legacy(args):
568 """Best-effort import of a Cora-style legacy state file. Liberal on input:
569 map what matches the known shapes, skip what doesn't, never fail."""
570 st, data = load_state(args.state)
571 if st == "corrupt":
572 return emit("CORRUPT")
573 if st == "absent":
574 data = new_state()
576 # Optional map from legacy channel id -> configured source id, so imported
577 # cursors land under the id the live connector reads via `cursor-get
578 # --source <config-id>`. Without it, a legacy "C42" cursor would be orphaned
579 # from a source configured as "slack-alpha" and the first sweep re-ingests
580 # everything. Absent or unparseable -> identity (legacy id used verbatim).
581 source_map = {}
582 if getattr(args, "source_map", None):
583 try:
584 parsed = json.loads(args.source_map)
585 if isinstance(parsed, dict):
586 source_map = {str(k): str(v) for k, v in parsed.items()}
587 except (ValueError, TypeError):
588 source_map = {}
590 legacy = _read_legacy(args.file)
591 cursors_imported = 0
592 items_imported = 0
593 if isinstance(legacy, dict):
594 cursors_imported = _import_channels(legacy, data, source_map)
595 items_imported = _import_legacy_items(legacy, data)
597 # Persist only when something changed: a fresh state file was seeded, or
598 # the import actually brought data in. A no-op import writes nothing.
599 if st == "absent" or cursors_imported or items_imported:
600 write_state(args.state, data)
601 return emit("OK", {
602 "cursors_imported": cursors_imported,
603 "items_imported": items_imported,
604 })
607def _read_legacy(path):
608 try:
609 with open(path, encoding="utf-8") as f:
610 raw = f.read()
611 except (OSError, UnicodeDecodeError):
612 return None
613 # Try JSON first (Cora persists JSON); fall back to our YAML subset.
614 try:
615 return json.loads(raw)
616 except ValueError:
617 pass
618 try:
619 return parse_document(raw)
620 except Exception:
621 return None
624def _import_channels(legacy, data, source_map=None):
625 channels = legacy.get("channels")
626 if not isinstance(channels, dict):
627 return 0
628 source_map = source_map or {}
629 sources = data.setdefault("sources", {})
630 count = 0
631 for chan_id, chan in channels.items():
632 if not isinstance(chan, dict):
633 continue
634 cursor = chan.get("last_processed_ts") or chan.get("cursor")
635 if cursor is None:
636 continue
637 source_id = source_map.get(str(chan_id), str(chan_id))
638 entry = sources.setdefault(source_id, {})
639 # Never regress an already-advanced cursor: a re-import against live
640 # state must not rewind a source to the legacy value and re-ingest
641 # (and re-acknowledge) everything since. Seed only when absent, or when
642 # the legacy value is not older than the current one.
643 current = entry.get("cursor")
644 if current is not None and _cursor_lt(str(cursor), str(current)):
645 continue
646 entry["cursor"] = cursor
647 count += 1
648 return count
651def _import_legacy_items(legacy, data):
652 raw_items = legacy.get("items")
653 items = data.setdefault("items", {})
654 count = 0
656 def add(item_id, fields):
657 if not item_id:
658 return 0
659 merged = dict(items.get(str(item_id), {}))
660 for k in ("status", "source", "channel", "title", "url"):
661 if k in fields and fields[k] is not None:
662 key = "source" if k == "channel" else k
663 merged.setdefault(key, fields[k])
664 items[str(item_id)] = merged
665 return 1
667 if isinstance(raw_items, dict):
668 for item_id, fields in raw_items.items():
669 if isinstance(fields, dict):
670 count += add(item_id, fields)
671 elif isinstance(raw_items, list):
672 for entry in raw_items:
673 if isinstance(entry, dict):
674 count += add(entry.get("id"), entry)
675 return count
678# --------------------------------------------------------------------------- #
679# CLI wiring
680# --------------------------------------------------------------------------- #
682def build_parser():
683 p = argparse.ArgumentParser(description="ce-sweep deterministic state engine")
684 sub = p.add_subparsers(dest="cmd", required=True)
686 def with_state(sp):
687 sp.add_argument("--state", required=True)
688 return sp
690 with_state(sub.add_parser("read"))
691 with_state(sub.add_parser("validate"))
693 up = with_state(sub.add_parser("upsert-item"))
694 up.add_argument("--id", required=True)
695 up.add_argument("--source", required=True)
696 up.add_argument("--json", required=True)
697 up.add_argument("--writer", required=True)
698 up.add_argument("--now")
700 cg = with_state(sub.add_parser("cursor-get"))
701 cg.add_argument("--source", required=True)
703 ca = with_state(sub.add_parser("cursor-advance"))
704 ca.add_argument("--source", required=True)
705 ca.add_argument("--to", required=True)
706 ca.add_argument("--past-item", required=True)
707 ca.add_argument("--writer", required=True)
708 ca.add_argument("--now")
710 la = with_state(sub.add_parser("lease-acquire"))
711 la.add_argument("--writer", required=True)
712 la.add_argument("--ttl-minutes", type=int, default=None)
713 la.add_argument("--now")
715 lr = with_state(sub.add_parser("lease-release"))
716 lr.add_argument("--writer", required=True)
718 rr = with_state(sub.add_parser("run-record"))
719 rr.add_argument("--writer", required=True)
720 rr.add_argument(
721 "--outcome", required=True,
722 choices=("completed", "aborted-locked", "partial", "failed"),
723 )
724 rr.add_argument("--counts", required=True)
725 rr.add_argument("--timestamp", required=True)
727 il = with_state(sub.add_parser("import-legacy"))
728 il.add_argument("--file", required=True)
729 il.add_argument(
730 "--source-map",
731 help='JSON object mapping legacy channel id -> configured source id, '
732 'e.g. \'{"C42":"slack-alpha"}\'. Absent -> legacy ids used verbatim.',
733 )
735 return p
738_HANDLERS = {
739 "read": cmd_read,
740 "validate": cmd_validate,
741 "upsert-item": cmd_upsert_item,
742 "cursor-get": cmd_cursor_get,
743 "cursor-advance": cmd_cursor_advance,
744 "lease-acquire": cmd_lease_acquire,
745 "lease-release": cmd_lease_release,
746 "run-record": cmd_run_record,
747 "import-legacy": cmd_import_legacy,
748}
750# Subcommands that read-modify-write the state file. The lease is a high-level
751# "who owns the sweep" guard, but some writes are deliberately lease-agnostic
752# (run-record for an aborted-locked run, validate, import-legacy). Two
753# concurrent invocations (an overlapping cron and manual sweep) could otherwise
754# interleave load -> mutate -> write and lose an update — e.g. an aborted run's
755# stale-snapshot write clobbering the holder's just-committed upsert. An OS
756# advisory lock held across each mutating RMW makes them mutually exclusive
757# regardless of lease ownership.
758_MUTATING = {
759 "validate", "upsert-item", "cursor-advance", "lease-acquire",
760 "lease-release", "run-record", "import-legacy",
761}
764def _run_locked(handler, args):
765 lock_path = str(args.state) + ".lock"
766 try:
767 lock_fd = open(lock_path, "w", encoding="utf-8")
768 except OSError:
769 return handler(args) # cannot create a lock file; degrade to unlocked
770 try:
771 fcntl.flock(lock_fd, fcntl.LOCK_EX)
772 return handler(args)
773 finally:
774 try:
775 fcntl.flock(lock_fd, fcntl.LOCK_UN)
776 finally:
777 lock_fd.close()
780def main(argv):
781 args = build_parser().parse_args(argv[1:])
782 handler = _HANDLERS[args.cmd]
783 try:
784 if _HAS_FCNTL and args.cmd in _MUTATING:
785 return _run_locked(handler, args)
786 return handler(args)
787 except Exception as exc: # never leak a traceback to the caller
788 sys.stderr.write(f"sweep-state: internal error: {exc}\n")
789 return emit("ERROR")
792if __name__ == "__main__":
793 sys.exit(main(sys.argv))