#!/usr/bin/env python3 """ claude_job.py — THE DRAFT-ENGINE SEAM: every pure text->JSON drafting turn in the pipeline lands here, and /draft-provider.json decides which engine answers it (see scripts/draft_provider.py; design + verified facts in docs/codex-port-design.md): claude-desktop-queue (default) enqueue to a file queue drained by the Claude Desktop s4l-worker scheduled task — the original design, and the only option when Claude Desktop is the box's sole authenticated engine. claude-p answer inline via the real `claude -p` (_run_claude_p; schema INLINED — the CLI parses --json-schema as JSON, a file path breaks it). codex-exec answer inline via the ChatGPT-app-bundled `codex exec` (_run_codex_exec; gpt-5.6-sol default, schema rides the prompt — OpenAI strict schema mode rejects our lenient schemas). gemini-api answer inline via the Gemini generateContent REST API (_run_gemini_api; gemini-pro-latest default with a one-shot flash downgrade on 404, schema rides the prompt for the same lenient-schema reason). The hosted-lane provider: key-authenticated, no local app, runs headless on Linux. All four return byte-identical claude `--output-format json` envelopes and the same 0/1/79 exit semantics, so callers can't tell them apart. The queue machinery below (next/result, worker notes, heartbeats) only runs for the default provider; the inline providers skip it entirely, which also means no warm-session leaks and no app-open requirement on those paths. The deterministic pipeline never calls `claude` directly; every invocation goes through scripts/run_claude.sh. For script tags mapped in TAG_TO_TYPE below, run_claude.sh delegates here instead of exec'ing the `claude` binary. The pipeline is otherwise untouched: it enqueues the same prompt + json-schema it would have passed to claude, blocks until a result appears, and gets back bytes shaped exactly like claude's `--output-format json` envelope, so the existing parsers don't change. Four roles: provider — (producer side, called by run_claude.sh) extract the prompt (stdin or trailing arg) + --json-schema, enqueue a typed job, BLOCK until a result lands, then print a claude-json-shaped envelope to stdout. next — (consumer side, called by a Claude Desktop scheduled task) atomically claim the oldest pending job of a given type and print it as JSON. result — (consumer side) store the JSON the task produced (validated) and unblock the waiting provider. Queue = plain files under /claude-queue/. No DB, no network. state_dir = $S4L_STATE_DIR or ~/.social-autoposter-mcp Job-type mapping is by run_claude.sh script_tag, and TAG_TO_TYPE is the ONLY router (the S4L_CLAUDE_PROVIDER env var was removed 2026-07-06): run_claude.sh asks `eligible --tag` and routes mapped tags through the queue unconditionally, on every machine. Only PURE text->JSON calls belong in the map; unmapped tags always run the real `claude -p`. Migrating a lane onto the queue = adding its tag to TAG_TO_TYPE, nothing else. """ from __future__ import annotations import argparse import json import os import re import signal import subprocess import sys import time import uuid # SAPS_->S4L_ env mirror (brand rename 2026-07-03): old launchd plists and # scheduled-task prompts still export SAPS_*; this process reads S4L_*. import s4l_env # noqa: E402 (lives next to this file in scripts/) s4l_env.mirror() # Draft-provider selection (2026-08-06): /draft-provider.json decides # which engine answers queue-eligible tags. Stamped once at setup, read here at # the seam; no env-var fallback layering. See scripts/draft_provider.py. import draft_provider # noqa: E402 (lives next to this file in scripts/) # Best-effort menu-bar activity narration. Importable because this script's own # directory (scripts/) is on sys.path[0] when run as `python3 .../claude_job.py`. # A failure to import (or to write) must NEVER affect the queue's real work. try: import s4l_activity as _activity # type: ignore except Exception: # pragma: no cover - cosmetic only _activity = None # script_tag -> queue type. ONLY pure text->JSON claude calls belong here. TAG_TO_TYPE = { "run-twitter-cycle-prep": "twitter-prep", "feedback-digest": "feedback-digest", # Topic-invention lane (queue-native since 2026-07-06; invent_topics.py # pins the provider to queue itself, no env switch). "invent-topic": "invent-topic", "invent-queries": "invent-queries", # Tail-link bridge (queue-native since 2026-07-06; moved from a post-time # call in twitter_post_plan.py to a draft-time call in # twitter_gen_links.py's Phase 2b-gen step, which already tolerates the # queue worker's cadence — see scripts/link_tail.py). "twitter-link-tail": "twitter-link-tail", # Reddit draft phase (queue-native since 2026-07-14): post_reddit.py's # draft turn became pure text->JSON (thread content is pre-fetched in # Python and inlined; no Bash tools), mirroring the twitter Phase 2b # tool-free conversion, so it rides the same universal worker. "post-reddit-draft": "reddit-draft", # Context mining (2026-07-31): scripts/context_mining.py inlines compacted # transcripts + the corpus into one pure text->JSON turn. Execution # guidance lives at the top of the prompt itself (not only in # TYPE_TO_WORKER_NOTES) because an installed-package worker claims with # ITS copy of this map, which may predate this type. "context-mining": "context-mining", } # queue type -> (activity state, label) the menu bar shows while the job is in # flight. Phase-2b prep is the reply drafting. Both the launchd provider (which # blocks for minutes) and the scheduled-task worker (which does the LLM turn) # narrate from this one map. TYPE_TO_ACTIVITY = { "twitter-prep": ("drafting", "draft"), "feedback-digest": ("learning", "feedback"), "invent-topic": ("learning", "new topic"), "invent-queries": ("learning", "new queries"), "twitter-link-tail": ("drafting", "link bridge"), # Same bare "draft" label as twitter-prep ON PURPOSE: the menu bar title # must stay concise (user rule 2026-07-14); the platform shows on the # review card itself, not in the tray. "reddit-draft": ("drafting", "draft"), "context-mining": ("learning", "mining context"), } # queue type -> execution notes PREPENDED to the prompt sidecar at claim time. # This keeps the scheduled-task worker fully type-blind: its SKILL.md is one # generic claim -> follow -> submit loop, and anything a specific job type # needs the executor to know (pacing, persist cadence) travels WITH the job. # The twitter-prep note exists because the host kills an unattended session # ~90s after its LAST tool call; drafting a whole batch in one silent turn # starves that clock (the v6 worker-prompt lesson, moved under the hood). TYPE_TO_WORKER_NOTES = { "twitter-prep": ( "WORKER EXECUTION NOTES (queue metadata; follow while executing the " "prompt below): this unattended session is terminated ~90 seconds after " "your LAST tool call. The prompt asks you to draft replies for SEVERAL " "candidates. Do NOT draft them all silently in one turn. Work ONE " "candidate at a time: draft its reply, then IMMEDIATELY run that " "candidate's log_draft.py persist command exactly as the prompt's " "persist step specifies (a quick Bash call), THEN move to the next. " "Those per-candidate Bash calls keep the session alive. Begin the first " "candidate promptly. Only after EVERY candidate is drafted and logged " "do you assemble and submit the single result JSON." ), "twitter-link-tail": ( "WORKER EXECUTION NOTES (queue metadata; follow while executing the " "prompt below): this job's answer is ONE LINE OF PLAIN TEXT, not a " "structured object — the prompt itself says 'no JSON, no markdown, no " "quotes'. Submit that single line as a bare JSON STRING (e.g. " "\"your one-line answer\"), not as an object like {\"text\": \"...\"} " "and not wrapped in any other keys. The caller unwraps a plain string " "result directly; wrapping it in an object will corrupt the reply." ), "reddit-draft": ( "WORKER EXECUTION NOTES (queue metadata; follow while executing the " "prompt below): this unattended session is terminated ~90 seconds " "after your LAST tool call. Every Reddit thread's content is already " "inlined in the prompt — never fetch or open a reddit.com URL; " "WebSearch/WebFetch are for EXTERNAL fact-checking only, per the " "prompt's THREAD CONTENT rules. Apply the prompt's SELECTION GATE to " "each candidate and work ONE thread at a time: draft both texts, " "then IMMEDIATELY run that thread's log_draft.py persist command " "exactly as the prompt's PERSIST step specifies (a quick Bash call), " "THEN move to the next. Those per-thread Bash calls keep the session " "alive. Only after EVERY candidate is handled do you assemble and " "submit ONE result object matching the schema: {\"posts\": [...], " "\"rejects\": [...]}. A candidate that fails the gate is simply " "absent from posts (add a rejects entry only for the structural " "false-positive cases the prompt describes)." ), "context-mining": ( "WORKER EXECUTION NOTES (queue metadata; follow while executing the " "prompt below): single-turn, tool-free job. Everything you need is " "inlined in the prompt (transcripts, corpus, prior decisions). Never " "use tools. Read, then submit ONE result object matching the schema: " "{\"proposals\": [...]}. An empty proposals list is a valid answer." ), } def _act_write(qtype: str) -> None: if _activity is None: return sl = TYPE_TO_ACTIVITY.get(qtype) if not sl: return try: _activity.write(sl[0], f"{sl[1]}…") except Exception: pass def _act_clear() -> None: if _activity is None: return try: _activity.clear() except Exception: pass def _fmt_dur(secs: float) -> str: """Compact human duration for the menu-bar label: '45s', '12m'.""" s = int(max(0, secs)) return f"{s}s" if s < 60 else f"{s // 60}m" def _act_write_progress( qtype: str, created: float, claimed_at: float | None, now: float ) -> None: """Granular in-flight menu-bar label, so a wedged cycle reads as the TRUTH instead of a static 'drafting replies' that lingers for the whole producer timeout (the failure mode where the worker never claims the job, or claims it and dies mid-run, looked identical to healthy drafting before this). - job still in pending/ (no worker has claimed it) -> ' ⧖' counting from enqueue. A growing '⧖18m' is the unmistakable tell that a scheduled-task worker is orphaned and nothing is draining. - job claimed (pending file gone -> moved to running/) -> ' ' counting from the claim, i.e. real drafting elapsed. The menu bar's stall watchdog parses the TRAILING 's'/'m' token out of this label (s4l_menubar._label_elapsed_secs), so the duration must stay the last number in the string whatever else changes. Purely cosmetic and best-effort: a write failure must never affect the queue.""" if _activity is None: return sl = TYPE_TO_ACTIVITY.get(qtype) if not sl: return state, base = sl if claimed_at is None: label = f"{base} ⧖{_fmt_dur(now - created)}" else: label = f"{base} {_fmt_dur(now - claimed_at)}" try: _activity.write(state, label) except Exception: pass # claude flags that consume the following argv token as their value, so the # value is never mistaken for the positional prompt. The CLI accepts BOTH # camelCase and kebab-case spellings for the tool filters; list both. Missing # kebab spellings bit on 2026-07-03: feedback_digest.py passes # "--disallowed-tools ", the parser treated it as boolean, the tools # list became the last positional, and _parse_claude_args returned it as the # prompt; every queue-routed digest job enqueued a tools list instead of the # real prompt and the worker rejected it (claude_failed=rc=1 hourly). VALUE_FLAGS = { "--mcp-config", "--json-schema", "--output-format", "--input-format", "--model", "--fallback-model", "--system-prompt", "--append-system-prompt", "--permission-mode", "--allowedTools", "--disallowedTools", "--allowed-tools", "--disallowed-tools", "--max-turns", "--add-dir", "--session-id", "--settings", } POLL_INTERVAL_S = 2.0 # Consumer-side poll cadence for `next --wait-seconds` (see cmd_next). Separate # constant from POLL_INTERVAL_S (the producer's own wait loop) because the two # sides have different cost profiles: the producer polls a single result file # it's blocked on anyway, while the worker's poll re-execs claude_job.py itself # each pass, so a slightly coarser cadence avoids needless process churn during # a multi-minute wait. WORKER_POLL_INTERVAL_S = 5.0 # Per-call budget the producer waits for ONE claude job (a query or a draft-prep # reasoning turn). Was 600s, which sat right at the edge of the draft call's real # ~9-10 min need: on the QA box ~41% of twitter-prep jobs breached 600s and got # dropped (each drop = a lost draft AND an orphaned over-running worker that # becomes a leaked agent-mode session). The DIRECT launchd `claude -p` lane has no # such per-call cap — its draft call just runs inside the 180-min cycle watchdog — # so 600s here made the queue lane diverge and silently fail where the direct lane # would not. 1800s (30 min) = 3x the real draft need, matching the sibling Twitter # engagement claude cap (engage-twitter Phase B gtimeout 1800), which removes the # drops while staying a bounded per-call value (not the whole-cycle budget). # COUPLING: reap_stale_claude_sessions.py reaps leaked workers at THIS value plus a # fixed margin (S4L_REAPER_AGE_MARGIN_SEC, default 300s) and MUST stay > it (a lower # reaper would SIGKILL a draft the producer is still waiting on). Both read # S4L_CLAUDE_QUEUE_TIMEOUT and both default to 1800; keep them in lockstep if you # change the base. DEFAULT_TIMEOUT_S = int(os.environ.get("S4L_CLAUDE_QUEUE_TIMEOUT", "1800")) # Jobs older than this (pending or running) are swept — a job nobody drained in # this long is a leftover from a timed-out producer or a dead worker, and keeping # it would feed a stale prompt to a worker much later. Default 2x the timeout. STALE_TTL_S = int(os.environ.get("S4L_CLAUDE_QUEUE_STALE_TTL", str(DEFAULT_TIMEOUT_S * 2))) # --------------------------------------------------------------------------- # EXPERIMENT (2026-06-29, per user): hide the "Top Posts by Project" few-shot # block from the Phase-2b drafting prompt. # # WHY: that block is ~42% of the prompt — top_performers.py emits up to 5 curated # example posts for EVERY project (~20 projects = ~74 examples, ~16k tokens), and # it is NOT scoped to the projects in this cycle's candidates. On a `.mcpb` box the # drafting runs inside a Claude Desktop scheduled-task session that the app # SIGTERMs after ~120s; the 38k-token prompt pushed the worker past that window # before it could submit, so jobs never drained. Dropping this block shrinks the # prompt to ~22k tokens (one Read, faster turns) while KEEPING the "Best Example # Per Style" block (which is style-scoped and tiny). # # This is a DELIVERY-LAYER trim only: the generator (top_performers.py, locked) is # untouched — we strip the section from the prompt text right before it is queued. # A loud marker is left in its place and a provider.log line is emitted, so it is # obvious the section was intentionally hidden, not lost. # # DEFAULT: ON (hidden). Set S4L_HIDE_TOP_BY_PROJECT=0 (or false/no) to restore the # full per-project example block. HIDE_TOP_BY_PROJECT = ( os.environ.get("S4L_HIDE_TOP_BY_PROJECT", "1").strip().lower() not in ("0", "false", "no", "off", "") ) def _strip_top_by_project(prompt: str) -> str: """Remove the '### Top Posts by Project' block from a drafting prompt. Returns the prompt with that one section replaced by a clearly-labelled HIDDEN marker (so anyone reading the prompt sees it was intentionally hidden behind S4L_HIDE_TOP_BY_PROJECT, not silently dropped). No-op if the block is absent (e.g. query prompts, or a report that produced no per-project posts). The section runs from its '### Top Posts by Project' header to the next '##'/ '###' header (normally '### Bottom N Posts'). """ start = "### Top Posts by Project" i = prompt.find(start) if i < 0: return prompt m = re.search(r"\n#{2,3} ", prompt[i + len(start):]) j = (i + len(start) + m.start() + 1) if m else len(prompt) marker = ( "### Top Posts by Project — HIDDEN\n" "[This per-project few-shot block (~16k tokens) was hidden at the delivery " "layer by claude_job.py via S4L_HIDE_TOP_BY_PROJECT (default ON, added " "2026-06-29) so the drafting worker fits Claude Desktop's ~120s " "scheduled-session window. Set S4L_HIDE_TOP_BY_PROJECT=0 to restore it. " "The 'Best Example Per Style' block above is kept.]\n\n" ) return prompt[:i] + marker + prompt[j:] # --------------------------------------------------------------------------- # # Queue layout # # --------------------------------------------------------------------------- # def _apply_state_dir_override(ns) -> None: """`--state-dir` wins over $S4L_STATE_DIR. The scheduled-task worker passes it explicitly so it always reads the SAME queue the launchd kicker writes to, regardless of what env the task session inherits.""" sd = getattr(ns, "state_dir", None) if sd: os.environ["S4L_STATE_DIR"] = sd def state_dir() -> str: return os.environ.get("S4L_STATE_DIR") or os.path.join( os.path.expanduser("~"), ".social-autoposter-mcp" ) def queue_root() -> str: return os.path.join(state_dir(), "claude-queue") def pending_dir(qtype: str) -> str: return os.path.join(queue_root(), "pending", qtype) def running_dir() -> str: return os.path.join(queue_root(), "running") def result_dir() -> str: return os.path.join(queue_root(), "result") def heartbeat_path() -> str: """Single file the worker stamps each time it claims or completes a job. Its mtime/contents prove the scheduled-task worker is actually draining the queue (vs. the SKILL.md merely existing on disk, which survives a Claude account switch and gave a false-green). Read by the MCP's autopilot liveness check and the stall detector. Empty-queue ("no jobs") fires deliberately do NOT stamp it — we want "is a job getting DRAINED", not "did a worker tick".""" return os.path.join(queue_root(), "worker-heartbeat.json") def _arm_deathwatch(job_id: str, qtype: str, batch: str) -> None: """Best-effort dead-man's-switch (2026-07-08): arm scripts/producer_deathwatch.py to flag an UNEXPECTED death (SIGKILL/OOM/hard crash) of THIS process while it's blocked in the cmd_provider() poll loop below — the exact gap that made orphaned salvage results ("worker drafted, no card") unexplainable. Every normal return path in cmd_provider() calls _disarm_deathwatch() first, so a clean exit never produces a report. Shared with run_claude.sh's direct-exec path (every non-queue platform), which calls producer_deathwatch.py's `arm`/`disarm` CLI directly instead of through this Python wrapper — see that file for the single implementation both callers share.""" try: import producer_deathwatch as pdw pdw.arm(os.getpid(), job_id, qtype, batch, call_path="queue") except Exception: pass def _disarm_deathwatch(job_id: str) -> None: try: import producer_deathwatch as pdw pdw.disarm(job_id) except Exception: pass def _stamp_heartbeat(event: str, qtype: str | None = None) -> None: """Best-effort: never let a heartbeat write failure break the queue.""" try: os.makedirs(queue_root(), exist_ok=True) _atomic_write( heartbeat_path(), {"at": time.time(), "event": event, "type": qtype or ""}, ) except Exception: pass def drain_status_path() -> str: """LATCHED autopilot-liveness marker the producer maintains: how many times in a row it has enqueued a job and timed out with NO worker draining it. Unlike a pending-job age check, this persists across the gaps between cycles (the producer removes the job on timeout, so there's no pending file to look at between cycles) — so the menu bar / dashboard / Sentry watcher can show a CONTINUOUS stall instead of one that flickers off every time a job is removed. Cleared (consecutive_timeouts=0) the moment a draft actually drains.""" return os.path.join(queue_root(), "drain-status.json") def _read_drain_status() -> dict: try: with open(drain_status_path()) as f: return json.load(f) except Exception: return {} def _mark_drain_success() -> None: """A job drained successfully -> clear the latched stall.""" try: os.makedirs(queue_root(), exist_ok=True) _atomic_write( drain_status_path(), {"consecutive_timeouts": 0, "last_success_at": time.time()}, ) except Exception: pass def _bump_drain_timeout() -> None: """The producer gave up waiting -> latch/escalate the stall.""" try: os.makedirs(queue_root(), exist_ok=True) cur = _read_drain_status() prev = int(cur.get("consecutive_timeouts", 0) or 0) cur["consecutive_timeouts"] = prev + 1 cur["last_timeout_at"] = time.time() _atomic_write(drain_status_path(), cur) except Exception: pass # --------------------------------------------------------------------------- # # Worker self-reap (2026-06-27; DEFAULT ON since 2026-07-04 / v1.6.202) # # --------------------------------------------------------------------------- # # A scheduled-task worker turn finishes its one queue iteration but Claude # Desktop keeps the agent-mode `claude` process warm (`--input-format # stream-json`), so finished workers pile up and leak RAM. The launchd reaper # (reap_stale_claude_sessions.py) is the GUARANTEE that bounds this. This # path is a faster, source-side trim: once THIS worker is provably done (no work # to do, or its result is already on disk), terminate OUR OWN session so it never # becomes part of the standing pool. It shipped opt-in (dormant unless # S4L_WORKER_SELF_REAP was set), which meant it ran nowhere; when Desktop # 1.18286.0 changed the cmdline shape the signature-fragile reaper missed too # and boxes leaked (Karol, 53 piled workers). Since 2026-07-04 it is ON by # default; opt OUT with S4L_WORKER_SELF_REAP=0 (see _self_reap_enabled). # # Safety properties: # * Respects the env kill switch (S4L_WORKER_SELF_REAP=0 disables it). # * Only ever targets a process in OUR OWN ancestry that matches the reaper's # worker signature (claude-code agent-mode session). The producer side # (run-twitter-cycle.sh -> python) has no such ancestor, so a misplaced call # there finds nothing and does nothing. # * Detached + delayed: a double-forked grandchild waits a few seconds (so the # current turn returns and prints its final line normally) before signalling. # * Re-verifies the target's cmdline right before SIGTERM, so a recycled PID is # never signalled. # * Best-effort throughout: never raises into the caller, never changes the # worker's exit code, never touches the result already written to disk. # LOOSE ancestry probe (2026-07-04). The original tuple duplicated the reaper's # strict cmdline signature, which Claude Desktop 1.18286.0 broke (Karol's second # leak: 53 workers piled up while both the reaper AND this self-reap failed the # same parse). Inside our OWN ancestry the strict fingerprint is unnecessary: # identification is DETERMINISTIC by construction — claude_job runs as a Bash # child of the session that invoked it, so the nearest claude-code agent-mode # ancestor IS that session. What the loose probe cannot tell apart is a # SCHEDULED WORKER session vs an INTERACTIVE session where someone ran # `claude_job next` by hand — that discrimination comes from the cwd gate in # _maybe_self_reap (worker tasks run in ~/.s4l-worker; interactive sessions # never do), not from cmdline shape. _SELF_REAP_SIG = ( "claude-code/", "--input-format stream-json", ) _S4L_WORKER_CWD = os.path.join(os.path.expanduser("~"), ".s4l-worker") def _self_reap_enabled() -> bool: # Default ON since 2026-07-04 (v1.6.202): the dormant flag meant the # source-side trim never ran anywhere, leaving the (signature-fragile) # launchd reaper as the only defense — and when Desktop changed the cmdline # shape, boxes leaked. Opt OUT with S4L_WORKER_SELF_REAP=0. return os.environ.get("S4L_WORKER_SELF_REAP", "").strip().lower() not in ( "0", "false", "no", "off", ) def _ps_pid_map() -> dict: """pid -> (ppid, command) for every process. Empty dict on any failure.""" out: dict = {} try: res = subprocess.run( ["/bin/ps", "-axo", "pid=,ppid=,command="], capture_output=True, text=True, timeout=10, ) except Exception: return out for line in res.stdout.splitlines(): parts = line.strip().split(None, 2) if len(parts) < 3: continue try: pid, ppid = int(parts[0]), int(parts[1]) except ValueError: continue out[pid] = (ppid, parts[2]) return out def _find_own_session(psmap: dict): """Walk OUR ancestry to the nearest claude-code agent-mode ancestor. Returns (pid, reverify_token) or None. The token is a stable cmdline prefix used to confirm the PID was not recycled before signalling — no UUID/path shape assumptions, so a Desktop cmdline change cannot blind this again.""" pid = os.getpid() seen: set = set() for _ in range(40): # bounded climb up the process tree if pid in seen: break seen.add(pid) ent = psmap.get(pid) if not ent: break ppid, cmd = ent if all(t in cmd for t in _SELF_REAP_SIG) and "Helpers/disclaimer" not in cmd: return pid, cmd[:160] pid = ppid if pid <= 1: break return None def _maybe_self_reap(delay: float = 6.0) -> None: """Terminate our own finished worker session. See block comment above. The cwd gate is the worker-vs-interactive discriminator: scheduled worker tasks run with cwd ~/.s4l-worker (enforced at task creation + menubar cwd rewrite), while an interactive/debug session that shells `claude_job` runs in a project dir. Interactive sessions are therefore never signalled, no matter what their cmdline looks like.""" if not _self_reap_enabled(): return try: cwd = os.getcwd() if cwd != _S4L_WORKER_CWD and not cwd.startswith(_S4L_WORKER_CWD + os.sep): return found = _find_own_session(_ps_pid_map()) if not found: return target_pid, token = found if os.fork() != 0: return # the worker continues + exits normally except Exception: return # child -> detach into its own session, then exit, orphaning the grandchild try: os.setsid() if os.fork() != 0: os._exit(0) except Exception: os._exit(0) # grandchild (fully detached): wait, re-verify, signal try: time.sleep(delay) cur = _ps_pid_map().get(target_pid) if cur and token in cur[1]: # same session, not a recycled PID try: os.kill(target_pid, signal.SIGTERM) except OSError: pass except Exception: pass os._exit(0) def _ensure_dirs(qtype: str | None = None) -> None: for d in (running_dir(), result_dir()): os.makedirs(d, exist_ok=True) if qtype: os.makedirs(pending_dir(qtype), exist_ok=True) def _atomic_write(path: str, obj) -> None: tmp = f"{path}.tmp.{os.getpid()}" with open(tmp, "w") as f: json.dump(obj, f) os.replace(tmp, path) def _atomic_write_text(path: str, text: str) -> None: tmp = f"{path}.tmp.{os.getpid()}" with open(tmp, "w") as f: f.write(text) os.replace(tmp, path) def _sweep_stale() -> int: """GC job files orphaned by dead producers. Returns count removed. Two different TTLs on purpose: - pending/: a job unclaimed past the producer's whole wait window (DEFAULT_TIMEOUT_S + margin) has no live producer left to consume its result — drafting it later only burns a worker turn on a stale prompt. This is the leak signature of a producer that was SIGKILLed by its external wrapper before its own timeout-cleanup ran (2026-07-18: three orphaned reddit-draft jobs piled up while paused). - running/: a claimed job may legitimately still be drafting, so it gets the longer STALE_TTL_S before we conclude the worker died too. """ removed = 0 now = time.time() pending_ttl = DEFAULT_TIMEOUT_S + 120 roots = [(running_dir(), STALE_TTL_S)] pend = os.path.join(queue_root(), "pending") if os.path.isdir(pend): roots += [(os.path.join(pend, d), pending_ttl) for d in os.listdir(pend)] for d, ttl in roots: if not os.path.isdir(d): continue for name in os.listdir(d): if not name.endswith(".json"): continue fp = os.path.join(d, name) try: with open(fp) as f: created = json.load(f).get("created_at", 0) if now - float(created) > ttl: os.remove(fp) removed += 1 except Exception: continue return removed # --------------------------------------------------------------------------- # # provider (producer side, run by run_claude.sh) # # --------------------------------------------------------------------------- # def _parse_claude_args(args: list[str]) -> tuple[str | None, str | None]: """Return (trailing_prompt, schema_path) from the verbatim claude argv.""" schema_path = None positionals: list[str] = [] i = 0 while i < len(args): a = args[i] if a == "--json-schema": schema_path = args[i + 1] if i + 1 < len(args) else None i += 2 continue if a in VALUE_FLAGS: i += 2 continue if a.startswith("-"): i += 1 # boolean flag (-p, --strict-mcp-config, --verbose, ...) continue positionals.append(a) i += 1 prompt = positionals[-1] if positionals else None return prompt, schema_path def _plog(msg: str) -> None: """Provider diagnostics go to a log file, NEVER stderr. The pipeline captures this wrapper's output with `2>&1` and parses the FIRST JSON value (raw_decode). Anything we print to stderr BEFORE the envelope (e.g. an "enqueued, waiting" line) lands ahead of the JSON and breaks the parse with "Expecting value: line 1 column 2". So stdout carries ONLY the final envelope and stderr stays silent; humans read provider.log instead. (fix 2026-06-24) """ try: os.makedirs(queue_root(), exist_ok=True) with open(os.path.join(queue_root(), "provider.log"), "a") as f: f.write(f"{time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())} pid={os.getpid()} {msg}\n") except Exception: pass def _emit_envelope(obj) -> None: """Print the claude `--output-format json` shaped envelope the pipeline's raw_decode + structured_output/result parsers expect, byte-compatible with both the queue path and a real claude run.""" envelope = { "type": "result", "subtype": "success", "is_error": False, "structured_output": obj, "result": json.dumps(obj) if not isinstance(obj, str) else obj, } sys.stdout.write(json.dumps(envelope)) sys.stdout.flush() def _run_codex_exec(ns, qtype: str, prompt: str, schema_text: str | None) -> int: """codex-exec provider: answer the turn inline via the ChatGPT-app-bundled `codex exec` (headless, subscription-authenticated). No queue, no worker, no app-open requirement. Same envelope, same 0/1/79 exit semantics as the queue path, so run_claude.sh callers can't tell the difference. No fallback to the queue on failure, by design: a broken codex login must fail loudly (rc=1) so the stall/deadman rails surface it, not silently strand jobs in a queue no worker drains. """ import tempfile codex = draft_provider.codex_bin() if not codex: _plog(f"codex-exec provider active but no codex binary found; failing {qtype}") return 1 batch = (os.environ.get("BATCH_ID") or os.environ.get("SA_CYCLE_ID") or "-").strip() or "-" job_id = uuid.uuid4().hex _arm_deathwatch(job_id, qtype, batch) _act_write(qtype) schema_path = None out_path = None try: out_fd, out_path = tempfile.mkstemp(prefix="s4l_codex_out_", suffix=".txt") os.close(out_fd) cmd = [ codex, "exec", "-", "-C", state_dir(), "-s", "read-only", "--skip-git-repo-check", "-o", out_path, ] # Deliberately NOT --output-schema: OpenAI's strict structured-output # mode rejects our lenient claude-style schemas (400 invalid_json_schema # unless additionalProperties:false + all-required everywhere). The # schema rides the prompt as guidance instead, mirroring the worker # rail, and _validate_against_schema stays the lenient gate. if schema_text: prompt = ( f"{prompt}\n\n" "FINAL ANSWER FORMAT: reply with ONLY a JSON value matching this " "JSON Schema. No prose, no markdown fences, nothing else:\n" f"{schema_text}" ) # Model default: gpt-5.6-sol (best 5.6 tier; user call 2026-08-06), env # override via S4L_CODEX_MODEL. Sol availability is plan-gated and the # app's models_cache.json under-reports it, so truth is probe-by-use: # on the specific "model not supported" 400 we retry ONCE with no -m # (the account's config default, e.g. gpt-5.6-terra) and log the # downgrade loudly. Any other failure stays loud with no retry. model = os.environ.get("S4L_CODEX_MODEL", "").strip() or "gpt-5.6-sol" # The user's global config may pin xhigh reasoning (fine for coding, # slow and token-hungry for a drafting turn). Default these turns to # medium; S4L_CODEX_REASONING overrides. effort = os.environ.get("S4L_CODEX_REASONING", "medium").strip() or "medium" cmd += ["-c", f'model_reasoning_effort="{effort}"'] budget = max(60, ns.timeout - 60) _plog(f"codex-exec start {qtype} job {job_id} batch={batch} bin={codex} model={model} effort={effort} timeout={budget}s") started = time.time() proc = None for attempt_model in (model, None): attempt_cmd = cmd + (["-m", attempt_model] if attempt_model else []) try: proc = subprocess.run( attempt_cmd, input=prompt, capture_output=True, text=True, timeout=max(60, budget - int(time.time() - started)), ) except subprocess.TimeoutExpired: _act_clear() _plog(f"codex-exec timed out after {budget}s on job {job_id} ({qtype})") return 79 # mirror the queue path's "blocked, skip cleanly" err_blob = f"{proc.stderr or ''}{proc.stdout or ''}" if ( attempt_model and "not supported when using Codex" in err_blob ): _plog( f"codex-exec model {attempt_model} unavailable on this account; " f"retrying job {job_id} with the config default model" ) continue break # Token usage rides stderr as "tokens used\nN"; best-effort log only. m = re.search(r"tokens used\s*\n?\s*([\d,]+)", proc.stderr or "") tokens = m.group(1) if m else "?" if proc.returncode != 0: tail = (proc.stderr or proc.stdout or "").strip()[-400:] _act_clear() _plog(f"codex-exec rc={proc.returncode} on job {job_id} ({qtype}); tail: {tail}") return 1 try: with open(out_path) as f: text = f.read().strip() except Exception as e: _act_clear() _plog(f"codex-exec produced no last-message file for job {job_id}: {e}") return 1 if not text: _act_clear() _plog(f"codex-exec empty answer on job {job_id} ({qtype})") return 1 # Parse the answer the same leniently the worker rail does: JSON when # it parses (objects for schema'd jobs, bare strings for link-tail), # fenced JSON unwrapped, otherwise the raw text as a plain string. obj = None parsed = False for candidate in (text, re.sub(r"^```(?:json)?\s*|\s*```$", "", text).strip()): try: obj = json.loads(candidate) parsed = True break except Exception: continue if not parsed: obj = text err = _validate_against_schema(obj, schema_text) if err: _act_clear() _plog(f"codex-exec result rejected for job {job_id} ({qtype}): {err}") return 1 _emit_envelope(obj) _mark_drain_success() _stamp_heartbeat("codex-exec", qtype) try: _ncand = len(obj.get("candidates")) if isinstance(obj, dict) and isinstance(obj.get("candidates"), list) else "?" except Exception: _ncand = "?" _plog( f"codex-exec done job {job_id} batch={batch} ({qtype}) in {int(time.time() - started)}s " f"tokens={tokens}; {_ncand} candidates -> producer assembles the plan" ) return 0 finally: for p in (schema_path, out_path): if p: try: os.remove(p) except OSError: pass _disarm_deathwatch(job_id) def _run_gemini_api(ns, qtype: str, prompt: str, schema_text: str | None) -> int: """gemini-api provider: answer the turn inline via the Gemini generateContent REST API. Key-authenticated, fully headless: this is the hosted-lane provider (no Claude Desktop, no ChatGPT app, runs on Linux). Same envelope, same 0/1/79 exit semantics as the other inline providers. Schema rides the prompt (not responseSchema): Gemini's structured-output schema dialect is an OpenAPI subset that rejects parts of our lenient claude-style schemas, so _validate_against_schema stays the single gate, matching the codex-exec precedent. responseMimeType application/json still forces bare-JSON output for schema'd jobs; schemaless jobs (link-tail) stay plain text. """ import urllib.error import urllib.request key = draft_provider.gemini_api_key() if not key: _plog(f"gemini-api provider active but no GEMINI_API_KEY / keychain gemini-api-key; failing {qtype}") return 1 batch = (os.environ.get("BATCH_ID") or os.environ.get("SA_CYCLE_ID") or "-").strip() or "-" job_id = uuid.uuid4().hex _arm_deathwatch(job_id, qtype, batch) _act_write(qtype) try: if schema_text: prompt = ( f"{prompt}\n\n" "FINAL ANSWER FORMAT: reply with ONLY a JSON value matching this " "JSON Schema. No prose, no markdown fences, nothing else:\n" f"{schema_text}" ) # Model default mirrors codex-exec's probe-by-use pattern: try the pro # alias first (drafting quality is the product), downgrade ONCE to the # flash alias on a model-not-found 404 and log it loudly. The -latest # aliases are Google-maintained so this never pins a dead model id. model = os.environ.get("S4L_GEMINI_MODEL", "").strip() or "gemini-pro-latest" fallback_model = "gemini-flash-latest" gen_config = {"temperature": 0.7, "maxOutputTokens": 16384} if schema_text: gen_config["responseMimeType"] = "application/json" body = json.dumps( { "contents": [{"role": "user", "parts": [{"text": prompt}]}], "generationConfig": gen_config, } ).encode() budget = max(60, ns.timeout - 60) _plog(f"gemini-api start {qtype} job {job_id} batch={batch} model={model} timeout={budget}s") started = time.time() resp_obj = None for attempt_model in (model, fallback_model if fallback_model != model else None): if not attempt_model: break url = ( "https://generativelanguage.googleapis.com/v1beta/models/" f"{attempt_model}:generateContent" ) req = urllib.request.Request( url, data=body, headers={"Content-Type": "application/json", "x-goog-api-key": key}, method="POST", ) try: with urllib.request.urlopen( req, timeout=max(60, budget - int(time.time() - started)) ) as resp: resp_obj = json.loads(resp.read().decode()) break except urllib.error.HTTPError as e: tail = "" try: tail = e.read().decode()[-400:] except Exception: pass if e.code == 404 and attempt_model == model: _plog( f"gemini-api model {attempt_model} not found (404); " f"retrying job {job_id} with {fallback_model}" ) continue _act_clear() _plog(f"gemini-api HTTP {e.code} on job {job_id} ({qtype}); tail: {tail}") return 1 except TimeoutError: _act_clear() _plog(f"gemini-api timed out after {budget}s on job {job_id} ({qtype})") return 79 # mirror the queue path's "blocked, skip cleanly" except Exception as e: _act_clear() _plog(f"gemini-api request failed on job {job_id} ({qtype}): {e}") return 1 if resp_obj is None: _act_clear() _plog(f"gemini-api no response on job {job_id} ({qtype})") return 1 try: cand = resp_obj["candidates"][0] text = "".join( p.get("text", "") for p in cand.get("content", {}).get("parts", []) ).strip() finish = cand.get("finishReason", "?") except (KeyError, IndexError, TypeError): _act_clear() block = (resp_obj.get("promptFeedback") or {}).get("blockReason", "?") _plog(f"gemini-api no candidates on job {job_id} ({qtype}); blockReason={block}") return 1 if not text: _act_clear() _plog(f"gemini-api empty answer on job {job_id} ({qtype}); finishReason={finish}") return 1 # Same lenient parse as codex-exec: JSON when it parses, fenced JSON # unwrapped, otherwise the raw text as a plain string (link-tail jobs). obj = None parsed = False for candidate in (text, re.sub(r"^```(?:json)?\s*|\s*```$", "", text).strip()): try: obj = json.loads(candidate) parsed = True break except Exception: continue if not parsed: obj = text err = _validate_against_schema(obj, schema_text) if err: _act_clear() _plog(f"gemini-api result rejected for job {job_id} ({qtype}): {err}") return 1 _emit_envelope(obj) _mark_drain_success() _stamp_heartbeat("gemini-api", qtype) usage = resp_obj.get("usageMetadata") or {} tokens = usage.get("totalTokenCount", "?") try: _ncand = len(obj.get("candidates")) if isinstance(obj, dict) and isinstance(obj.get("candidates"), list) else "?" except Exception: _ncand = "?" _plog( f"gemini-api done job {job_id} batch={batch} ({qtype}) in {int(time.time() - started)}s " f"tokens={tokens} finish={finish}; {_ncand} candidates -> producer assembles the plan" ) return 0 finally: _disarm_deathwatch(job_id) def _run_claude_p(ns, qtype: str, prompt: str, schema_text: str | None) -> int: """claude-p provider: answer the turn inline via the real `claude -p` (Claude Code CLI, subscription-authenticated). No queue, no worker, no Desktop-app requirement. stdout passes through untouched: it IS the claude json envelope the callers parse. Key transform vs the caller's original argv: the schema goes INLINE (the CLI parses the --json-schema value as JSON; a file path 400s), and the prompt rides stdin so huge drafting prompts never hit ARG_MAX.""" batch = (os.environ.get("BATCH_ID") or os.environ.get("SA_CYCLE_ID") or "-").strip() or "-" job_id = uuid.uuid4().hex _arm_deathwatch(job_id, qtype, batch) _act_write(qtype) cmd = ["claude", "-p", "--output-format", "json"] if schema_text: cmd += ["--json-schema", schema_text] budget = max(60, ns.timeout - 60) _plog(f"claude-p start {qtype} job {job_id} batch={batch} timeout={budget}s") started = time.time() try: proc = subprocess.run(cmd, input=prompt, capture_output=True, text=True, timeout=budget) except subprocess.TimeoutExpired: _act_clear() _plog(f"claude-p timed out after {budget}s on job {job_id} ({qtype})") _disarm_deathwatch(job_id) return 79 except FileNotFoundError: _act_clear() _plog(f"claude-p provider active but no `claude` CLI on PATH; failing {qtype}") _disarm_deathwatch(job_id) return 1 sys.stdout.write(proc.stdout or "") sys.stdout.flush() if proc.returncode == 0 and '"subtype":"success"' in (proc.stdout or ""): _mark_drain_success() _stamp_heartbeat("claude-p", qtype) _plog(f"claude-p done job {job_id} batch={batch} ({qtype}) in {int(time.time() - started)}s") else: _act_clear() tail = (proc.stderr or "").strip()[-300:] _plog(f"claude-p rc={proc.returncode} on job {job_id} ({qtype}); stderr tail: {tail}") _disarm_deathwatch(job_id) return proc.returncode def cmd_provider(ns) -> int: _apply_state_dir_override(ns) qtype = TAG_TO_TYPE.get(ns.tag) if not qtype: # Not a queue-eligible call. Exit non-zero so run_claude.sh's caller # treats it as a claude failure and runs its own fallback path. _plog(f"tag '{ns.tag}' is not queue-eligible; no provider") return 1 stdin_text = "" if not sys.stdin.isatty(): try: stdin_text = sys.stdin.read() except Exception: stdin_text = "" trailing_prompt, schema_path = _parse_claude_args(ns.claude_args) prompt = stdin_text if stdin_text.strip() else (trailing_prompt or "") if not prompt.strip(): _plog("empty prompt; nothing to enqueue") return 1 schema_text = None if schema_path and os.path.exists(schema_path): try: with open(schema_path) as f: schema_text = f.read() except Exception: schema_text = None # Delivery-layer trim: hide the "Top Posts by Project" block from drafting # prompts so the worker fits the ~120s scheduled-session window. See # HIDE_TOP_BY_PROJECT / _strip_top_by_project above. Drafting jobs only. if qtype == "twitter-prep" and HIDE_TOP_BY_PROJECT: _before_len = len(prompt) prompt = _strip_top_by_project(prompt) if len(prompt) != _before_len: _plog( "hid 'Top Posts by Project' block: -%d chars " "(S4L_HIDE_TOP_BY_PROJECT on; set =0 to restore)" % (_before_len - len(prompt)) ) # S4L paused: refuse to enqueue at all. Workers won't claim while paused, so # an enqueued job would just sit in pending/ while this producer burns its # whole wait window "drafting" and then dies to its external timeout. # Applies to every provider: paused means no drafting, period. if _is_paused(): _plog(f"S4L paused; refusing to enqueue {qtype} job") return 1 # Provider branch (2026-08-06). codex-exec and claude-p answer inline; # anything else is the queue (the Claude Desktop worker default). _prov = draft_provider.get() if _prov == "codex-exec": return _run_codex_exec(ns, qtype, prompt, schema_text) if _prov == "claude-p": return _run_claude_p(ns, qtype, prompt, schema_text) if _prov == "gemini-api": return _run_gemini_api(ns, qtype, prompt, schema_text) job_id = uuid.uuid4().hex created = time.time() # Cycle batch id (run-twitter-cycle.sh exports BATCH_ID) so every provider.log # line correlates to the cycle's own twitter-cycle-.log. '-' when off-cycle. batch = (os.environ.get("BATCH_ID") or os.environ.get("SA_CYCLE_ID") or "-").strip() or "-" _ensure_dirs(qtype) _sweep_stale() # clear leftovers from prior timed-out producers before enqueuing job = { "job_id": job_id, "type": qtype, "tag": ns.tag, "prompt": prompt, "schema": schema_text, "created_at": created, } # Filename is _.json so a plain sorted() listing is FIFO. fname = f"{int(created * 1e9):020d}_{job_id}.json" pending_path = os.path.join(pending_dir(qtype), fname) running_path = os.path.join(running_dir(), fname) _atomic_write(pending_path, job) _plog(f"enqueued {qtype} job {job_id} batch={batch}; waiting for a scheduled task (timeout {ns.timeout}s)") _arm_deathwatch(job_id, qtype, batch) # Narrate the (multi-minute) block to the menu bar. The launchd draft lane has # no other activity writer, so without this the box looks idle while it works. # Cleared by run-draft-and-publish.sh's exit trap at cycle end (and by the # worker's cmd_result), so we deliberately do NOT clear on the success path # here — that would flicker the indicator off between the cycle's claude calls. _act_write(qtype) res_path = os.path.join(result_dir(), f"{job_id}.json") # Give up before ns.timeout, not at it: external wrappers (post_reddit's # subprocess timeout, gtimeout in shell lanes) start their clocks at # process start — before stdin read + enqueue — so an internal deadline of # exactly created + timeout LOSES that race and the producer gets SIGKILLed # before the cleanup below ever runs (2026-07-18 deathwatch "unexpected # death" at enqueue+1800s, leaking pending files and a stuck "drafting" # label). The 60s margin guarantees the graceful path wins. deadline = created + max(60, ns.timeout - 60) last_hb = created # last menu-bar heartbeat (see below) claimed_at = None # set the moment a worker moves the job pending/ -> running/ while time.time() < deadline: now = time.time() # Pause landed mid-wait: abandon immediately with the same cleanup the # timeout path does, instead of blocking the full wait window against a # queue no worker will drain while paused. if _is_paused(): _plog(f"job {job_id} abandoned: S4L paused mid-wait; cleaning up") for p in (pending_path, running_path): try: os.remove(p) except OSError: pass _act_clear() _disarm_deathwatch(job_id) return 1 # A worker claims a job by atomically renaming pending/ -> running/, so the # pending file vanishing is our signal that drafting actually STARTED (vs. # the job still sitting unclaimed). Latch the claim time once so the label # can distinguish "waiting for a worker" from "worker is drafting" and show # the right elapsed for each. if claimed_at is None and not os.path.exists(pending_path): claimed_at = now # Heartbeat the menu-bar label so its `since` stays fresh for the whole # multi-minute block. The consumer (s4l_state.read_activity) ages a label # out after a TTL, so without this refresh a long drafting turn would look # stale and the spinner would wrongly blink to idle. Refreshing here means # the label is fresh EXACTLY while real work is happening, and stops the # instant we return or die — so the consumer's TTL can then expire it # instead of it freezing forever. Throttled to ~10s; best-effort only. The # label now carries claim-state + elapsed so a stuck cycle reads honestly # ("queued 18m") instead of a reassuring static "drafting replies". if now - last_hb >= 10: _act_write_progress(qtype, created, claimed_at, now) last_hb = now if os.path.exists(res_path): try: with open(res_path) as f: res = json.load(f) except Exception: time.sleep(POLL_INTERVAL_S) continue os.remove(res_path) if res.get("status") == "error": _plog(f"job {job_id} returned error: {res.get('error', 'unknown')}") _disarm_deathwatch(job_id) return 1 obj = res.get("result") # Emit a claude `--output-format json` shaped envelope so the # pipeline's existing raw_decode + structured_output/result parser # is byte-compatible. envelope = { "type": "result", "subtype": "success", "is_error": False, "structured_output": obj, "result": json.dumps(obj) if not isinstance(obj, str) else obj, } sys.stdout.write(json.dumps(envelope)) sys.stdout.flush() # A worker drained this job -> the autopilot is alive; clear any latched # stall so the menu bar / dashboard / Sentry watcher recover. _mark_drain_success() # Success-consume event: the SILENT gap that made an orphaned result # (worker wrote it, producer died before consuming) indistinguishable # from a healthy one. A result file only survives in result/ if it was # NEVER consumed (we os.remove above), so "consumed" here + a surviving # file = the orphan signature the salvage reconciler keys on. try: _ncand = len(obj.get("candidates")) if isinstance(obj, dict) and isinstance(obj.get("candidates"), list) else "?" except Exception: _ncand = "?" _plog(f"consumed result for job {job_id} batch={batch} ({qtype}); {_ncand} candidates -> producer assembles the plan") _disarm_deathwatch(job_id) return 0 time.sleep(POLL_INTERVAL_S) # Don't leak the job: remove it from pending/running so it can't be drafted # later with a stale prompt (and so /tmp doesn't accumulate stuck jobs). for p in (pending_path, running_path): try: os.remove(p) except OSError: pass # We gave up waiting for a worker — drop the "drafting" menu-bar label we kept # re-asserting while blocked. Otherwise it lingers and the menu bar shows # "drafting replies" forever (masking the autopilot-stalled ⚠) even though no # routine ever claimed the job. _act_clear() # Latch the stall so it persists across the gap until the next cycle enqueues # (no pending file exists between cycles, so an instantaneous queue check would # flicker the ⚠ off). Cleared only when a draft actually drains. _bump_drain_timeout() _plog(f"timed out after {ns.timeout}s waiting for job {job_id} batch={batch} ({qtype}); removed the job") _disarm_deathwatch(job_id) return 79 # mirror run_claude.sh's "blocked, skip cleanly" exit code # --------------------------------------------------------------------------- # # next (consumer side, run by a scheduled task) # # --------------------------------------------------------------------------- # def _agent_session_pid(): """Best-effort: the Claude agent-mode SESSION pid running THIS worker — the exact process the stale-session reaper (reap_stale_claude_sessions.py) would target. We climb our own process tree to the ancestor whose cmd carries the reaper's worker signature ('claude-code/' + 'local-agent-mode-sessions') and return its pid, so the claim can be stamped with it and the reaper can SPARE that session for the whole drafting turn (instead of SIGTERMing it at the short grace window — the 2026-06-29 draft-kill regression). None if not identifiable; the reaper then falls back to its newest-spare heuristic. """ try: out = subprocess.run( ["/bin/ps", "-axo", "pid=,ppid=,command="], capture_output=True, text=True, timeout=10, ).stdout info = {} for line in out.splitlines(): m = re.match(r"\s*(\d+)\s+(\d+)\s+(.*)$", line) if m: info[int(m.group(1))] = (int(m.group(2)), m.group(3)) pid = os.getpid() for _ in range(16): # bounded climb up the tree ent = info.get(pid) if not ent or ent[0] <= 1: break ppid = ent[0] pcmd = info.get(ppid, (0, ""))[1] if ("claude-code/" in pcmd) and ("local-agent-mode-sessions" in pcmd): return ppid pid = ppid except Exception: return None return None def _attempt_claim(ns, qtype: str) -> bool: """One pass over the pending dirs: try to claim the oldest job. Returns True (and prints the claimed job's payload) iff a job was claimed; False if the queue was empty this pass. Split out of cmd_next so the poll loop below can call it repeatedly without duplicating the claim/stamp/print logic.""" # "any" (the universal type-blind worker) scans EVERY pending type dir; # a comma list scans those types; a single type keeps legacy behavior. # Job filenames start with a zero-padded nanosecond timestamp, so one # global lexicographic sort is oldest-first across types. if qtype == "any": _ensure_dirs() root = os.path.join(queue_root(), "pending") try: scan_types = sorted( d for d in os.listdir(root) if os.path.isdir(os.path.join(root, d)) ) except FileNotFoundError: scan_types = [] else: scan_types = [t.strip() for t in qtype.split(",") if t.strip()] for t in scan_types: _ensure_dirs(t) entries = [] for t in scan_types: pend = pending_dir(t) try: for name in os.listdir(pend): entries.append((name, pend)) except FileNotFoundError: continue entries.sort(key=lambda e: e[0]) for name, pend in entries: if not name.endswith(".json") or name.endswith(".tmp"): continue src = os.path.join(pend, name) dst = os.path.join(running_dir(), name) try: os.rename(src, dst) # atomic claim; loser of a race gets FileNotFound except FileNotFoundError: continue try: with open(dst) as f: job = json.load(f) except Exception: continue # Stamp the agent-session pid that holds THIS claim so the reaper spares it # for the whole drafting turn (see _agent_session_pid above). agent_pid = _agent_session_pid() if agent_pid: job["claim_pid"] = agent_pid job["claimed_at"] = time.time() prompt_file = None schema_file = None if ns.prompt_file: prompt_file = os.path.join(queue_root(), f"prompt-{job['job_id']}.md") prompt_body = job.get("prompt") or "" # Per-type execution notes ride in the sidecar, not the worker # prompt, so the worker stays type-blind (see TYPE_TO_WORKER_NOTES). notes = TYPE_TO_WORKER_NOTES.get(job.get("type") or "") if notes: prompt_body = f"{notes}\n\n---\n\n{prompt_body}" _atomic_write_text(prompt_file, prompt_body) job["prompt_file"] = prompt_file schema = job.get("schema") if schema: schema_file = os.path.join(queue_root(), f"schema-{job['job_id']}.json") _atomic_write_text(schema_file, schema) job["schema_file"] = schema_file # ALWAYS persist the claim back (claim_pid + any prompt/schema sidecars) so # the reaper can read claim_pid; previously this only happened on the # --prompt-file lane, leaving claim_pid unstamped for inline callers. _atomic_write(dst, job) _plog( f"claimed {job.get('type') or qtype} job {job['job_id']}; " + (f"agent-session pid={agent_pid} stamped (reaper will spare it)" if agent_pid else "agent-session pid NOT found (reaper falls back to newest-spare)") ) # Narrate the scheduled-task worker's drafting turn to the menu bar. This # is the lane that actually runs the LLM; it persists until cmd_result # clears it (or the kicker's exit trap does). Covers the box's autopilot. _act_write(job.get("type") or qtype) # Liveness pulse: a routine actually claimed a job. Proves the worker is # firing, not just that its SKILL.md exists (see heartbeat_path()). _stamp_heartbeat("claim", job.get("type") or qtype) # Hand the consumer exactly what it needs to do the work and report back. payload = {"job_id": job["job_id"], "type": job["type"]} if ns.prompt_file: payload["prompt_file"] = prompt_file payload["schema_file"] = schema_file else: payload["prompt"] = job["prompt"] payload["schema"] = job.get("schema") print(json.dumps(payload)) return True return False def _is_paused() -> bool: # S4L paused (mcp/src/index.ts pauseS4L / the menubar pause button): the # one flag file every pipeline surface honors. return os.path.exists(os.path.join(state_dir(), "paused.flag")) def cmd_next(ns) -> int: _apply_state_dir_override(ns) # GC jobs orphaned by dead/killed producers BEFORE claiming (and before the # pause check, so orphans left behind by a pause don't sit until the next # enqueue-side sweep — while paused nothing enqueues, so this is the only # sweeper running). Without this, a resume would hand workers stale jobs # whose producer died long ago. swept = _sweep_stale() if swept: _plog(f"swept {swept} stale queue job file(s) before claim pass") # S4L paused: report no work instead of claiming a job, so an # already-queued draft can't drain/post while paused even though the # scheduled task worker still fires on its own cadence. if _is_paused(): print(json.dumps({})) return 0 qtype = ns.type wait_seconds = max(0, ns.wait_seconds or 0) # wait_seconds=0 (default) is the legacy single-shot behavior: one pass, # then done. wait_seconds>0 polls in a bounded loop within THIS ONE process # (one Bash call from the calling session's perspective) instead of relying # on the scheduled task's own cron cadence to re-check. A finished-but-empty # pass sleeps WORKER_POLL_INTERVAL_S and tries again until the deadline. # # COUPLING: the reaper's claim_grace (S4L_REAPER_CLAIM_GRACE_SEC in # reap_stale_claude_sessions.py) must stay >= whatever --wait-seconds the # worker prompt actually passes, plus margin — a claimless session polling # inside this loop is legitimate, not a husk, and a too-tight claim_grace # would SIGTERM it mid-poll before it ever gets a chance to claim. deadline = time.time() + wait_seconds start = time.time() attempt = 0 while True: # Re-check pause on every iteration, not just at entry: a worker # already inside its wait window when Pause is clicked must stop # claiming immediately (observed 2026-07-18: pause at 12:24, a # pre-pause poller claimed a reddit-draft job at 12:29 anyway). if _is_paused(): _plog(f"polling {qtype}: paused.flag appeared mid-wait; exiting without claiming") break if _attempt_claim(ns, qtype): return 0 attempt += 1 remaining = deadline - time.time() if remaining <= 0: break # Observability for the polling path (2026-07-07): production has no # per-attempt visibility otherwise — only the final claim gets a _plog # line, so there was no way to see a worker actually waiting out a real # gap vs. just getting lucky on its first check. Silent for the legacy # wait_seconds=0 single-shot path (never had this concern, no new noise). _plog( f"polling {qtype}: attempt {attempt}, {time.time() - start:.0f}s elapsed, " f"no job yet ({remaining:.0f}s left in wait window)" ) time.sleep(min(WORKER_POLL_INTERVAL_S, remaining)) print(json.dumps({})) # no work found within the wait window _maybe_self_reap() # idle turn, no job claimed — safe to retire this session return 0 # --------------------------------------------------------------------------- # # result (consumer side, run by a scheduled task) # # --------------------------------------------------------------------------- # def _validate_against_schema(obj, schema_text: str | None) -> str | None: """Lenient validation. Returns an error string or None if acceptable. We deliberately avoid a jsonschema dependency (not guaranteed on the box). Enforce only what matters: the result is a JSON object and carries the schema's top-level required keys. The prompt itself describes the full shape. """ if schema_text: try: schema = json.loads(schema_text) except Exception: schema = None if isinstance(schema, dict): if schema.get("type") == "object" and not isinstance(obj, dict): return "result must be a JSON object" required = schema.get("required") if isinstance(required, list) and isinstance(obj, dict): missing = [k for k in required if k not in obj] if missing: return f"result missing required keys: {missing}" return None def cmd_result(ns) -> int: _apply_state_dir_override(ns) _ensure_dirs() # The worker's drafting turn ends here (success or failure); drop the menu-bar # label so nothing lingers. The provider's next enqueue re-asserts the right # label for the cycle's following claude call, if any. _act_clear() # Liveness pulse: a routine completed a drain. Keeps the heartbeat fresh # across the whole claim->result span the worker was alive. _stamp_heartbeat("result") job_id = ns.job # Read the produced result (JSON object) from a file or stdin. if ns.result_file and ns.result_file != "-": with open(ns.result_file) as f: raw = f.read() else: raw = sys.stdin.read() raw = raw.strip() running = None # Locate the claimed job to recover its schema (filename carries job_id). schema_text = None cleanup_files: list[str] = [] try: for name in os.listdir(running_dir()): if name.endswith(f"_{job_id}.json"): with open(os.path.join(running_dir(), name)) as f: job = json.load(f) schema_text = job.get("schema") cleanup_files = [ p for p in (job.get("prompt_file"), job.get("schema_file")) if p ] running = os.path.join(running_dir(), name) break except FileNotFoundError: running = None if ns.error: _atomic_write( os.path.join(result_dir(), f"{job_id}.json"), {"status": "error", "error": raw or "unspecified"}, ) if running and os.path.exists(running): os.remove(running) for p in cleanup_files: try: os.remove(p) except OSError: pass print(f"[claude_job] recorded error for job {job_id}", file=sys.stderr) _maybe_self_reap() # error recorded, turn done — safe to retire this session return 0 try: obj = json.loads(raw) except Exception as e: print( f"[claude_job] result for job {job_id} is not valid JSON: {e}", file=sys.stderr, ) return 2 err = _validate_against_schema(obj, schema_text) if err: print(f"[claude_job] result for job {job_id} rejected: {err}", file=sys.stderr) return 3 _atomic_write( os.path.join(result_dir(), f"{job_id}.json"), {"status": "done", "result": obj}, ) if running and os.path.exists(running): os.remove(running) for p in cleanup_files: try: os.remove(p) except OSError: pass print(f"[claude_job] stored result for job {job_id}", file=sys.stderr) _maybe_self_reap() # result delivered to disk — safe to retire this session return 0 def cmd_eligible(ns: argparse.Namespace) -> int: """Routing probe for run_claude.sh: exit 0 when the tag is queue-mapped, 1 otherwise. TAG_TO_TYPE is the single routing truth — no env var. Provider selection happens INSIDE cmd_provider (codex-exec and claude-p both run inline there); eligible stays a pure tag probe. A fall-through design for claude-p was tried and reverted same-day 2026-08-06: pipeline callers pass --json-schema as a FILE PATH, and the real `claude` CLI parses that value as inline JSON, so the fall-through broke on every schema'd tag. The inline runner rewrites the schema to inline form. """ return 0 if ns.tag in TAG_TO_TYPE else 1 def main() -> int: p = argparse.ArgumentParser(description="claude -p queue shim") sub = p.add_subparsers(dest="cmd", required=True) pe = sub.add_parser("eligible", help="exit 0 iff --tag is queue-mapped (router probe)") pe.add_argument("--tag", required=True) pe.set_defaults(func=cmd_eligible) pp = sub.add_parser("provider", help="enqueue + block-poll (run by run_claude.sh)") pp.add_argument("--tag", required=True) pp.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT_S) pp.add_argument("--state-dir", default=None, help="override $S4L_STATE_DIR") pp.add_argument("claude_args", nargs=argparse.REMAINDER) pp.set_defaults(func=cmd_provider) pn = sub.add_parser("next", help="claim oldest pending job of a type") pn.add_argument("--type", required=True) pn.add_argument("--state-dir", default=None, help="override $S4L_STATE_DIR") pn.add_argument( "--prompt-file", action="store_true", help="write the prompt/schema to sidecar files and print their paths", ) pn.add_argument( "--wait-seconds", type=int, default=0, help="poll for a job up to this many seconds before giving up " "(0 = legacy single-shot: check once and return immediately)", ) pn.set_defaults(func=cmd_next) pr = sub.add_parser("result", help="store a job's result") pr.add_argument("--job", required=True) pr.add_argument("--result-file", default="-", help="path to JSON, or - for stdin") pr.add_argument("--error", action="store_true", help="record a failure") pr.add_argument("--state-dir", default=None, help="override $S4L_STATE_DIR") pr.set_defaults(func=cmd_result) ns = p.parse_args() # argparse.REMAINDER keeps a leading "--"; drop it. if getattr(ns, "claude_args", None) and ns.claude_args and ns.claude_args[0] == "--": ns.claude_args = ns.claude_args[1:] return ns.func(ns) if __name__ == "__main__": sys.exit(main())