on_scheduled hook: operator-defined post-schedule command

Runs after every successful schedule with the result as JSON on stdin;
failures are surfaced but never undo the schedule. Silent on dry runs and
failed schedules (tested). Keeps channel-specific tracking glue out of
the core - the origin channel's vault integration becomes a hook script
instead of a fork.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: Dyane681

README.md

@@ -70,6 +70,15 @@ than guessing.
 **Scheduling only, never live.** A badly timed scheduled post can be deleted; a live
 post cannot. Going live is deliberately not exposed — reversibility first.
 
+**Extensible where workflows differ, opinionated where they don't.** Every operator
+has their own tracking (a wiki to stamp, a review folder to archive, a content
+database to update). That glue stays out of the core: set `on_scheduled` in the
+config to any command and it runs after each successful schedule with the result as
+JSON on stdin (`{video, slot, post_id, series, caption, media_url}`). Hook failures
+are surfaced in the result but never undo the schedule. Security note: the hook is
+an arbitrary command sourced from your config file — keep `config.json` writable
+only by you.
+
 ## Tested invariants
 
 `tests/test_invariants.py` (plain pytest, fake API, zero network) pins the promises

docs/case-study.md

@@ -76,6 +76,13 @@ Numbers from the live channel (as of 2026-07-16):
   mature. The double-count guard on the series cap was found by the demo script
   in this repo, then pinned by a test.
 
+- **Channel glue belongs in a hook, not a fork.** The original pipeline stamps three
+  wiki files, archives the review copy and feeds a learning ledger after every
+  schedule. None of that belongs in a generic tool — but all of it fits one
+  extension point. That observation became the `on_scheduled` hook: the channel's
+  entire vault integration can run as a ~20-line hook script consuming the JSON
+  payload, instead of maintaining a diverging fork of the scheduler.
+
 ## Next iteration
 
 - Feed per-platform totals into the planner as platforms diverge (the config

examples/config.example.json

@@ -26,5 +26,6 @@
     "ig_user_id": "1784...",
     "fb_page_id": "1140..."
   },
-  "planner_sources": []
+  "planner_sources": [],
+  "on_scheduled": ""
 }

postpeer_pilot/config.py

@@ -39,6 +39,12 @@
     # performance sources
     "tiktok_handle": "",          # enables the yt-dlp puller when set
     "meta": {},                   # {"env": "~/.config/meta/.env", "ig_user_id": "...", "fb_page_id": "..."}
+    # post-schedule hook: shell command run after EVERY successful schedule, with the
+    # result as JSON on stdin ({video, slot, post_id, series, caption, media_url}).
+    # Wire your own tracking/archiving here. SECURITY: this executes an arbitrary
+    # command from the config file — it is an operator decision, keep config.json
+    # writable only by you. Empty = disabled. Hook failures never undo the schedule.
+    "on_scheduled": "",
 }
 
 

postpeer_pilot/scheduler.py

@@ -8,10 +8,12 @@
 Scheduling only — going live immediately is deliberately NOT offered here; a wrongly
 timed scheduled post can be deleted, a live post cannot.
 """
+import json
+import subprocess
 from datetime import date
 from pathlib import Path
 
-from . import api, plan
+from . import api, config, plan
 
 
 def _caption(video: Path, caption: str | None) -> str:
@@ -30,6 +32,22 @@ def _title(video: Path, title: str | None) -> str | None:
     return sidecar.read_text().strip() if sidecar.exists() else None
 
 
+def _fire_hook(payload: dict) -> dict | None:
+    """Run the operator's on_scheduled hook (config) with the schedule result as JSON
+    on stdin. Fire-and-report: a failing hook is surfaced in the result but NEVER
+    undoes or fails the schedule itself — the post is already placed."""
+    cmd = config.load().get("on_scheduled", "")
+    if not cmd:
+        return None
+    try:
+        p = subprocess.run(cmd, shell=True, input=json.dumps(payload, ensure_ascii=False),
+                           capture_output=True, text=True, timeout=120)
+        return {"ok": p.returncode == 0,
+                **({"stderr": p.stderr[-300:]} if p.returncode != 0 else {})}
+    except Exception as e:
+        return {"ok": False, "stderr": f"{type(e).__name__}: {e}"}
+
+
 def _already_scheduled(video: Path) -> str | None:
     """Idempotency guard: has THIS file already been scheduled into a future slot?
     Retrying a batch after a partial failure must not double-post the successes."""
@@ -76,11 +94,16 @@ def schedule(videos: list, captions: list | None = None, titles: list | None = N
                             "orphaned_upload": url, "error": f"{type(e).__name__}: {e}"})
             continue
         ok = bool(r.get("id") or r.get("success") or r.get("postId"))
+        result = {"ok": ok, "video": video.name, "slot": slot, "response": r}
         if ok:
-            plan.record(slot, video.name, series,
-                        str(r.get("postId") or r.get("id") or ""), caption=cap)
+            pid = str(r.get("postId") or r.get("id") or "")
+            plan.record(slot, video.name, series, pid, caption=cap)
+            hook = _fire_hook({"video": str(video), "slot": slot, "post_id": pid,
+                               "series": series, "caption": cap, "media_url": url})
+            if hook is not None:
+                result["hook"] = hook
         else:
             taken.pop(slot, None)          # post failed -> slot is still free
-            r = {**r, "orphaned_upload": url}
-        results.append({"ok": ok, "video": video.name, "slot": slot, "response": r})
+            result["response"] = {**r, "orphaned_upload": url}
+        results.append(result)
     return results

tests/test_hook.py

@@ -0,0 +1,43 @@
+"""on_scheduled hook: fires with the right payload, never breaks scheduling, silent
+on dry runs and failures-to-schedule."""
+import json
+from datetime import date, timedelta
+
+from postpeer_pilot import scheduler
+
+START = date.today() + timedelta(days=30)
+
+
+def _set_hook(home, cmd):
+    cfg = json.loads((home / "config.json").read_text())
+    cfg["on_scheduled"] = cmd
+    (home / "config.json").write_text(json.dumps(cfg))
+
+
+def test_hook_receives_payload(home, fake_api, video):
+    out = home / "hook_payload.json"
+    _set_hook(home, f"cat > {out}")
+    r = scheduler.schedule([video("hooked")], series="s1", start=START)
+    assert r[0]["ok"] and r[0]["hook"]["ok"]
+    payload = json.loads(out.read_text())
+    assert payload["slot"] == r[0]["slot"]
+    assert payload["post_id"] and payload["series"] == "s1"
+    assert "Caption for hooked" in payload["caption"]
+
+
+def test_failing_hook_does_not_fail_schedule(home, fake_api, video):
+    _set_hook(home, "exit 7")
+    r = scheduler.schedule([video("h2")], start=START)
+    assert r[0]["ok"] is True                      # schedule stands
+    assert r[0]["hook"]["ok"] is False             # failure surfaced, not swallowed
+    assert fake_api.creates == 1
+
+
+def test_hook_silent_on_dry_run_and_failure(home, fake_api, video):
+    out = home / "should_not_exist"
+    _set_hook(home, f"touch {out}")
+    scheduler.schedule([video("h3")], dry_run=True, start=START)
+    assert not out.exists()                        # dry run: no hook
+    fake_api.fail_creates = 99
+    r = scheduler.schedule([video("h4")], start=START)
+    assert r[0]["ok"] is False and not out.exists()  # failed schedule: no hook