From ec663a93c844843892c93b2368cedcffab659df5 Mon Sep 17 00:00:00 2001 From: Godopu Date: Thu, 24 Sep 2026 13:37:05 +0000 Subject: [PATCH] feat(issue-34): a job records its parent; submit --parent-job (MAM_JOB_ID by default) and the brief tells the agent to pass it on - registry: parent_job field, register --parent-job fails closed on an unknown parent, children_of/descendants_of, list --parent - wrapper: --parent-job; $MAM_JOB_ID default (ignored with a warning when this registry lacks it); brief line and one sentence at the end of the injected instructions - create_session: the onboarding instructions carry the same sentence Co-Authored-By: Claude Opus 5.5 --- .agents/skills/lib_py/delegation/registry.py | 48 +++++++++++++++++++ .../scripts/create_session.sh | 3 +- .../multi-agent-mux-delegate-job | 33 +++++++++++-- 3 files changed, 78 insertions(+), 6 deletions(-) diff --git a/.agents/skills/lib_py/delegation/registry.py b/.agents/skills/lib_py/delegation/registry.py index 72975ef..983cb50 100755 --- a/.agents/skills/lib_py/delegation/registry.py +++ b/.agents/skills/lib_py/delegation/registry.py @@ -89,6 +89,7 @@ def register_job( attempt_id: int = 1, transport: str = "herdr", supersede_older_attempt: bool = False, + parent_job: Optional[str] = None, ) -> str: """Create a new ``pending`` job record and return its id. @@ -102,6 +103,9 @@ def register_job( case it passes the orchestrator's so the events it publishes carry the identity the orchestrator's subscriber expects. ``transport`` records how the prompt reaches the agent (``herdr`` pane or ``remote`` dispatch topic). + + Issue #34: ``parent_job`` names the job this one was delegated from, so a + loop waiting on the parent can find and wait for it. None is a root job. """ job_id = job_id or generate_job_id(bits) heartbeat_sec = max(1, int(heartbeat_sec)) @@ -145,6 +149,8 @@ def register_job( "heartbeat_misses": heartbeat_misses, "ack_timeout_sec": max(1, int(ack_timeout_sec)), "transport": transport, + # Issue #34: the job this one was delegated from (None = a root job) + "parent_job": parent_job, } with registry_lock(registry_dir): if mqtt_common._job_path(job_id, registry_dir).exists(): @@ -245,6 +251,36 @@ def list_jobs(registry_dir: str = DEFAULT_REGISTRY_DIR, status: Optional[str] = return records +def children_of(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> List[Dict[str, Any]]: + """Issue #34: the records whose ``parent_job`` is ``job_id``, oldest first.""" + return [r for r in list_jobs(registry_dir) if r.get("parent_job") == job_id] + + +def descendants_of(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> List[Dict[str, Any]]: + """Issue #34: every job delegated from ``job_id``, directly or through its + children, breadth first. Each record gets a ``depth`` (1 = a child). One + scan of the registry; a record is visited once, so a cycle cannot loop.""" + by_parent: Dict[str, List[Dict[str, Any]]] = {} + for r in list_jobs(registry_dir): + if r.get("parent_job"): + by_parent.setdefault(r["parent_job"], []).append(r) + out: List[Dict[str, Any]] = [] + seen = {job_id} + frontier, depth = [job_id], 0 + while frontier: + depth += 1 + nxt = [] + for pid in frontier: + for r in by_parent.get(pid, []): + if r.get("job_id") in seen: + continue + seen.add(r["job_id"]) + out.append(dict(r, depth=depth)) + nxt.append(r["job_id"]) + frontier = nxt + return out + + def append_event(job_id: str, registry_dir: str, payload: Dict[str, Any]) -> None: """Append one event payload as a JSON line to the job's events log. Best effort, debug-only; failures are logged but never raised to the caller.""" @@ -261,6 +297,7 @@ def append_event(job_id: str, registry_dir: str, payload: Dict[str, Any]) -> Non __all__ = [ "register_job", "pick_pending", "update_status", "load_job", "list_jobs", "append_event", "generate_job_id", "get_feedback", + "children_of", "descendants_of", ] @@ -353,9 +390,12 @@ def _build_parser() -> argparse.ArgumentParser: p_reg.add_argument("--reviewer", default=None) p_reg.add_argument("--reviewer-session", default=None) p_reg.add_argument("--max-iterations", type=int, default=5) + p_reg.add_argument("--parent-job", default=None, + help="the job this one is delegated from (Issue #34); must exist in this registry") p_list = sub.add_parser("list", help="list jobs (optionally by status)") p_list.add_argument("--status", default=None) + p_list.add_argument("--parent", default=None, help="only the direct children of this job") p_list.add_argument("--json", action="store_true") p_get = sub.add_parser("get", help="print one job record as JSON") @@ -497,6 +537,11 @@ def main(argv: Optional[List[str]] = None) -> int: rd = args.registry_dir if args.command == "register": + # Issue #34: a mistyped parent must not silently make a root job - the + # loop would then not wait for it, which is the bug this field fixes. + if args.parent_job and not mqtt_common._job_path(args.parent_job, rd).exists(): + print(f"unknown parent job {args.parent_job}", file=sys.stderr) + return 1 job_id = register_job( prompt=_prompt_from_args(parser, args), agent=args.agent, @@ -519,12 +564,15 @@ def main(argv: Optional[List[str]] = None) -> int: dispatch_token=args.dispatch_token, attempt_id=args.attempt_id, job_id=args.job_id, + parent_job=args.parent_job, ) print(job_id) return 0 if args.command == "list": records = list_jobs(rd, status=args.status) + if args.parent: + records = [r for r in records if r.get("parent_job") == args.parent] if args.json: print(json.dumps(records, ensure_ascii=False, indent=2)) else: diff --git a/.agents/skills/multi-agent-mux-create/scripts/create_session.sh b/.agents/skills/multi-agent-mux-create/scripts/create_session.sh index 0547594..bd62e27 100755 --- a/.agents/skills/multi-agent-mux-create/scripts/create_session.sh +++ b/.agents/skills/multi-agent-mux-create/scripts/create_session.sh @@ -642,7 +642,8 @@ On progress (optional): $pub --event progress --detail ''. On success run: $pub --event completed --detail ''. On failure run: $pub --event error --detail ''. -Task: read the complete task in $task_brief and follow it." +Task: read the complete task in $task_brief and follow it. +If you delegate any part of it with multi-agent-mux-delegate-job submit, add --parent-job $DELEGATE_JOB_ID so the orchestrator waits for it." # Inject instructions into the herdr pane rc=0 diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index d67ffc3..c3ff26c 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job @@ -98,11 +98,12 @@ _pub_cmd() { } # _instructions [iteration]: the prompt injected into the agent (direct -# and loop jobs share it; test18 pins parts of this wording). +# and loop jobs share it; test18 pins parts of this wording, so Issue #34's +# sub-delegation sentence goes at the end). _instructions() { local pub="$1" iter="${2:-}" brief="$REGISTRY_DIR/$JOB_ID/brief.md" - printf 'Your job_id is "%s". Detailed task requirements, instructions, and target output paths%s are documented in the task brief file at: %s. Please READ and follow %s to complete your work. Commands: start='"'"'%s --event started --heartbeat %s'"'"' (run it FIRST, within %s s; it also starts an automatic heartbeat timer), success='"'"'%s --event completed --detail '"'"', error='"'"'%s --event error --detail '"'"'.' \ - "$JOB_ID" "${iter:+ for iteration $iter}" "$brief" "$brief" "$pub" "$HEARTBEAT_SEC" "$ACK_TIMEOUT" "$pub" "$pub" + printf 'Your job_id is "%s". Detailed task requirements, instructions, and target output paths%s are documented in the task brief file at: %s. Please READ and follow %s to complete your work. Commands: start='"'"'%s --event started --heartbeat %s'"'"' (run it FIRST, within %s s; it also starts an automatic heartbeat timer), success='"'"'%s --event completed --detail '"'"', error='"'"'%s --event error --detail '"'"'. If you delegate any part of this work with multi-agent-mux-delegate-job submit, add --parent-job %s so the orchestrator waits for it.' \ + "$JOB_ID" "${iter:+ for iteration $iter}" "$brief" "$brief" "$pub" "$HEARTBEAT_SEC" "$ACK_TIMEOUT" "$pub" "$pub" "$JOB_ID" } usage() { @@ -115,7 +116,11 @@ multi-agent-mux-delegate-job [options] [--registry-dir ] [--dry-run] [--role ] [--type ] [--reviewer ] [--reviewer-session ] [--max-iterations ] - [--counterpart-role ] [--strict-role-check] + [--counterpart-role ] [--strict-role-check] [--parent-job ] + # --parent-job names the job this one is delegated from, so a loop + # waiting on that job waits for this one too (Issue #34). It must + # exist in the registry. Default: $MAM_JOB_ID (set by the remote + # runner), ignored with a warning when the registry lacks it. # --prompt-file passes the prompt by file: one argv string is capped # at 131,072 bytes on Linux (Issue #36). The file is read, not kept. # --agent-session herdr: injects into a local herdr pane; @@ -144,6 +149,7 @@ COUNTERPART_ROLE="$DEFAULT_COUNTERPART_ROLE" STRICT_ROLE_CHECK=0 ROLE_ALIASES_JSON='{"worker": ["worker", "creator"], "planner": ["planner"], "reviewer": ["reviewer"]}' COUNTERPART_ROLE_EXPLICIT=0 +PARENT_JOB="" parse_opts() { while [[ $# -gt 0 ]]; do @@ -169,6 +175,7 @@ parse_opts() { --max-iterations) MAX_ITERATIONS="$2"; shift 2;; --counterpart-role) COUNTERPART_ROLE="$2"; COUNTERPART_ROLE_EXPLICIT=1; shift 2;; --strict-role-check) STRICT_ROLE_CHECK=1; shift;; + --parent-job) PARENT_JOB="$2"; shift 2;; *) echo "unknown option: $1" >&2; usage; exit 1;; esac done @@ -204,6 +211,20 @@ cmd_submit() { _abs_registry_dir LOGS_DIR_ABS="$(_logs_dir_abs 2>/dev/null || true)" + # Issue #34: an explicit --parent-job fails closed in the registry; the + # $MAM_JOB_ID default may name a job of another host's registry (a remote + # mirror), so it is only used when this registry has it. + local reg_parent=() + if [[ -n "$PARENT_JOB" ]]; then + reg_parent=(--parent-job "$PARENT_JOB") + elif [[ -n "${MAM_JOB_ID:-}" ]]; then + if [[ -f "$REGISTRY_DIR/$MAM_JOB_ID.json" ]]; then + PARENT_JOB="$MAM_JOB_ID"; reg_parent=(--parent-job "$PARENT_JOB") + else + echo "WARNING: MAM_JOB_ID=$MAM_JOB_ID is not in $REGISTRY_DIR; registering without a parent job" >&2 + fi + fi + # 1) register job (prints the new job id) JOB_ID="$("$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" register \ "${reg_prompt[@]}" --agent "$AGENT" --agent-session "$AGENT_SESSION" --role "$DELEGATE_ROLE" \ @@ -211,7 +232,7 @@ cmd_submit() { --heartbeat-sec "$HEARTBEAT_SEC" --heartbeat-misses "$HEARTBEAT_MISSES" --ack-timeout "$ACK_TIMEOUT" \ --transport "$(_is_remote "$AGENT_SESSION" && echo remote || echo herdr)" \ --job-type "$TYPE" --reviewer "$REVIEWER" --reviewer-session "$REVIEWER_SESSION" \ - --max-iterations "$MAX_ITERATIONS")" + --max-iterations "$MAX_ITERATIONS" ${reg_parent[@]+"${reg_parent[@]}"})" echo "registered job: $JOB_ID" # 1-1) Provision job directory and write direct brief.md (MAM Job Restructuring) @@ -227,6 +248,7 @@ cmd_submit() { - **Timeout**: $TIMEOUT s (Idle: $IDLE_TIMEOUT s) - **Liveness contract**: heartbeat every $HEARTBEAT_SEC s (automatic once you publish \`started --heartbeat $HEARTBEAT_SEC\`), unresponsive after $HEARTBEAT_MISSES missed heartbeats, \`started\` expected within $ACK_TIMEOUT s - **Output Report Path**: $REGISTRY_DIR/$JOB_ID/$AGENT-reports/report-final.md +- **Parent for sub-delegations**: pass \`--parent-job $JOB_ID\` to \`multi-agent-mux-delegate-job submit\` ## 🔎 Task Description $PROMPT @@ -402,6 +424,7 @@ EOF - **Iteration**: $iteration - **Liveness contract**: heartbeat every $HEARTBEAT_SEC s (automatic once you publish \`started --heartbeat $HEARTBEAT_SEC\`), unresponsive after $HEARTBEAT_MISSES missed heartbeats, \`started\` expected within $ACK_TIMEOUT s - **Output Report Path**: $REGISTRY_DIR/$JOB_ID/${clean_session}-reports/report-final.md +- **Parent for sub-delegations**: pass \`--parent-job $JOB_ID\` to \`multi-agent-mux-delegate-job submit\` ## 🔎 Task Description $current_prompt