"""v3.6 Runtime Harness — Layer 1: Dispatch Marker Writer. chair_authorization_id=CHAIR-AUTH-TASK-2704-V36-CONTROL-PLANE-P0-MVP-260528 Contract: - Write a dispatch marker JSON to memory/events/.dispatched--.json immediately after a successful dispatch. - 7 required fields: schedule_id, executor, bot_key_hash, fire_time, prompt_sha, task_md_sha_before, dispatch_method - bot_key_hash = sha256(bot_key).hexdigest()[:16] — raw key NEVER stored - prompt_sha = sha256(prompt.encode('utf-8')).hexdigest() - dispatch_method enum: dispatch_py | direct_cron_bot_key | fallback_safety_net - Safe-fail: ALL exceptions silently swallowed. Returns None on failure. Never raises. """ # chair_authorization_id propagation signature formalized in task-2707 (task-2705+1 lineage) from __future__ import annotations import hashlib import json import os from datetime import datetime, timezone from typing import Optional MARKER_VERSION = "v1" MARKER_TYPE = "dispatch_marker" # Resolved relative to workspace root at runtime _DEFAULT_EVENTS_DIR = "/home/jay/workspace/memory/events" DISPATCH_METHODS = frozenset({"dispatch_py", "direct_cron_bot_key", "fallback_safety_net"}) _REQUIRED_FIELDS = frozenset({ "schedule_id", "executor", "bot_key_hash", "fire_time", "prompt_sha", "task_md_sha_before", "dispatch_method", }) def write_dispatch_marker( task_id: str, schedule_id: Optional[str], executor: str, bot_key: str, fire_time: str, prompt: str, task_md_sha_before: Optional[str], dispatch_method: str, events_dir: Optional[str] = None, team_suffix: Optional[str] = None, chair_authorization_id: Optional[str] = None, ) -> Optional[dict]: """Write a dispatch marker JSON file. Args: task_id: Task identifier (e.g. "task-2704"). schedule_id: Cron schedule ID returned by dispatch. executor: Team name or bot persona string. bot_key: Raw bot API key — hashed before storage; never persisted raw. fire_time: ISO 8601 dispatch time string. prompt: Full prompt text; stored as SHA-256 hash only. task_md_sha_before: SHA-256 of task markdown at dispatch entry; None if unavailable. dispatch_method: One of "dispatch_py" | "direct_cron_bot_key" | "fallback_safety_net". events_dir: Override directory for marker files (default: memory/events/). team_suffix: Optional team suffix for filename deduplication. Returns: dict with marker contents on success, None on any failure. CRITICAL: This function is absolutely safe-fail. Any exception — including bad argument types, filesystem errors, import errors — is silently swallowed. Callers' dispatch flow is NEVER affected. """ try: return _write_dispatch_marker_impl( task_id=task_id, schedule_id=schedule_id, executor=executor, bot_key=bot_key, fire_time=fire_time, prompt=prompt, task_md_sha_before=task_md_sha_before, dispatch_method=dispatch_method, events_dir=events_dir, team_suffix=team_suffix, chair_authorization_id=chair_authorization_id, ) except Exception: # Absolute safe-fail: never propagate any exception to caller return None def _write_dispatch_marker_impl( task_id: str, schedule_id: Optional[str], executor: str, bot_key: str, fire_time: str, prompt: str, task_md_sha_before: Optional[str], dispatch_method: str, events_dir: Optional[str], team_suffix: Optional[str], chair_authorization_id: Optional[str], ) -> Optional[dict]: """Internal implementation — may raise; outer wrapper catches all exceptions.""" # Validate dispatch_method if dispatch_method not in DISPATCH_METHODS: # Log but don't raise — use "dispatch_py" as safe default dispatch_method = "dispatch_py" # Compute hashes bot_key_hash = hashlib.sha256(bot_key.encode("utf-8")).hexdigest()[:16] prompt_sha = hashlib.sha256(prompt.encode("utf-8")).hexdigest() # Build timestamp for filename now_utc = datetime.now(timezone.utc) ts_for_file = now_utc.strftime("%Y%m%d") ts_iso = now_utc.isoformat() # Build filename dir_path = events_dir or _DEFAULT_EVENTS_DIR if team_suffix: filename = f"{task_id}.dispatched-{team_suffix}-{ts_for_file}.json" else: filename = f"{task_id}.dispatched-{ts_for_file}.json" # If file already exists with a different timestamp portion, use microsecond suffix marker_path = os.path.join(dir_path, filename) if os.path.exists(marker_path): ts_us = now_utc.strftime("%Y%m%d%H%M%S%f") if team_suffix: filename = f"{task_id}.dispatched-{team_suffix}-{ts_us}.json" else: filename = f"{task_id}.dispatched-{ts_us}.json" marker_path = os.path.join(dir_path, filename) # Build marker document marker = { "marker_version": MARKER_VERSION, "marker_type": MARKER_TYPE, "task_id": task_id, "schedule_id": schedule_id, "executor": executor, "bot_key_hash": bot_key_hash, "fire_time": fire_time, "prompt_sha": prompt_sha, "task_md_sha_before": task_md_sha_before, "dispatch_method": dispatch_method, "written_at": ts_iso, "chair_authorization_id": chair_authorization_id, } # Validate required fields (self-check) missing = _REQUIRED_FIELDS - {k for k, v in marker.items() if v is not None or k == "task_md_sha_before"} if missing: # Do not raise — return None if schema integrity check fails return None # Atomic write os.makedirs(dir_path, exist_ok=True) tmp_path = marker_path + ".tmp" try: with open(tmp_path, "w", encoding="utf-8") as fh: json.dump(marker, fh, ensure_ascii=False, indent=2) os.replace(tmp_path, marker_path) finally: # Clean up tmp on failure if os.path.exists(tmp_path): try: os.unlink(tmp_path) except Exception: pass # Emit a JSONL decision record (safe-fail — do not let logger failure affect us) try: from scripts.harness.v36.runtime_decision_logger import log_runtime_event log_runtime_event( event_type="marker_written", task_id=task_id, payload={ "marker_path": marker_path, "schedule_id": schedule_id, "dispatch_method": dispatch_method, "bot_key_hash": bot_key_hash, }, ) except Exception: pass return marker