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 <noreply@anthropic.com>
This commit is contained in:
1 parent
7a4a0c2091
commit
ec663a93c8
3 files changed
+78
-6
No files matched your search
@@ -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:
|
||||
|
||||
Reference in new issue
Block a user