docs: update DRAFT_PLAN.md with problem definitions & gRPC justification, move agent rules to .agents/
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
# multi-agent-mux-delegate-job 스킬
|
||||
|
||||
작업(Job)을 자율 에이전트(claude-code/codex/opencode/human)에게 위임하고 MQTT
|
||||
이벤트 채널로 비동기 관찰하는 Hermes 스킬. **시작점은 [`SKILL.md`](./SKILL.md).**
|
||||
작업(Job)을 자율 에이전트(claude-code/hermes/agy/cline/codex/opencode/human)에게 위임하고 MQTT
|
||||
이벤트 채널로 비동기 관찰하는 범용 에이전트 협업 스킬. **시작점은 [`SKILL.md`](./SKILL.md).**
|
||||
|
||||
- 프로토콜/스키마: [`job-protocol.md`](./job-protocol.md)
|
||||
- 브로커 PoC→운영 전환: [`mqtt-broker-setup.md`](./mqtt-broker-setup.md)
|
||||
|
||||
@@ -1,385 +1,94 @@
|
||||
---
|
||||
name: multi-agent-mux-delegate-job
|
||||
description: "Delegate a unit of work to any autonomous agent (claude-code, codex, opencode, or a human) and observe it asynchronously over an MQTT event channel. Each job gets a unique id, a registry record (prompt, broker, status, timeouts), and a single per-job topic that carries started/permission_required/progress/completed/error events as schema-versioned JSON. The delegator starts a subscriber first, runs the agent, and treats a completed/error event or a timeout as the job's terminal state. Ships a working reference implementation (publish_event.py, job_subscriber.py, registry.py, mqtt_common.py, multi-agent-mux-delegate-job wrapper) plus a PoC-to-production path: validate on a public broker, then move to an authenticated TLS broker by changing config only — no code change. Use when you need fire-and-observe delegation, multi-job fan-out across tmux sessions, or a uniform completion-signal protocol shared by several agent types."
|
||||
version: 1.0.0
|
||||
author: Hermes Agent
|
||||
description: "Delegate a unit of work to any autonomous agent (claude-code, hermes, agy, cline, codex, or a human) and observe it asynchronously over an MQTT event channel. Supported roles include orchestrator, worker, and reviewer."
|
||||
version: 1.1.0
|
||||
author: Multi-Agent System
|
||||
license: MIT
|
||||
platforms: [linux, macos, windows]
|
||||
metadata:
|
||||
hermes:
|
||||
tags: [agent-delegation, mqtt, jobs, orchestration, async-completion]
|
||||
related_skills: [claude-code, codex, opencode, hermes-agent-skill-authoring]
|
||||
---
|
||||
|
||||
# multi-agent-mux-delegate-job — Async Job Delegation over MQTT
|
||||
|
||||
Delegate a unit of work to an autonomous agent, then **observe** it instead of
|
||||
blocking on it. Every job gets a unique id and a registry record; the agent
|
||||
publishes lifecycle events (`started`, `permission_required`, `progress`,
|
||||
`completed`, `error`) to a per-job MQTT topic; the delegator subscribes and
|
||||
treats `completed`/`error` — or a timeout — as the terminal state.
|
||||
Delegate a unit of work to any autonomous agent, then **observe** it asynchronously instead of blocking. Every job gets a unique ID and a registry record. The worker agent publishes lifecycle events (`started`, `permission_required`, `progress`, `completed`, `error`) to a per-job MQTT topic, and the delegator/orchestrator subscribes to verify the final state.
|
||||
|
||||
This skill is a **reference implementation**: copy the files in this directory
|
||||
into your project and customise. The `communication_over_mqtt` project is the
|
||||
canonical concrete instance.
|
||||
This skill allows any agent (`claude-code`, `hermes`, `agy`, `cline`, etc.) to play any role: **Orchestrator/Delegator**, **Worker/Implementer**, or **Reviewer**.
|
||||
|
||||
## Overview
|
||||
---
|
||||
|
||||
The model is deliberately small. A **job** is one delegated task. An **agent**
|
||||
is a worker (a claude-code tmux session, a codex run, a human). The **registry**
|
||||
(`.mam/jobs/<id>.json`) holds everything about a job so nothing important
|
||||
lives in environment variables — which means one tmux session can process many
|
||||
jobs sequentially, and many sessions can fan out in parallel, with no env
|
||||
collisions. The **event channel** is one MQTT topic per job carrying JSON
|
||||
payloads; `event` discriminates the type.
|
||||
## Roles in Multi-Agent Mux
|
||||
|
||||
Responsibility is split into exactly one entry point each:
|
||||
[`publish_event.py`](./scripts/publish_event.py) emits events (registry lookup,
|
||||
monotonic `seq`, retry+backoff) and [`job_subscriber.py`](./scripts/job_subscriber.py)
|
||||
observes them (timeouts, terminal state machine, defensive parsing). Shared
|
||||
logic lives in [`mqtt_common.py`](./scripts/mqtt_common.py); registry I/O in
|
||||
[`registry.py`](./scripts/registry.py). The demo `publisher.py`/`subscriber.py`
|
||||
in the host project stay frozen.
|
||||
- **Orchestrator (Delegator)**: Initiates the job, coordinates other agents, handles loops and reviews, and commits final changes.
|
||||
- **Worker (Implementer)**: Receives the brief file or task prompt, performs the implementation, and emits started/completed/error events.
|
||||
- **Reviewer**: Evaluates git diffs or artifacts produced by the worker, and responds with a `completed` event containing `"PASS"` or feedback.
|
||||
|
||||
Two stages, same code. **PoC** runs on the public `broker.hivemq.com` to wire up
|
||||
the protocol. **Production** moves to your own authenticated TLS broker — the
|
||||
switch is **config only** (env vars + the registry `broker.*` block), never a
|
||||
code change. See [`mqtt-broker-setup.md`](./mqtt-broker-setup.md).
|
||||
---
|
||||
|
||||
## When to Use / When NOT to Use
|
||||
## Core Commands (CLI)
|
||||
|
||||
**Use when:**
|
||||
- you want **fire-and-observe** delegation — kick off work and get a completion
|
||||
signal rather than blocking a terminal;
|
||||
- several agent types (claude-code, codex, opencode, human) must follow **one**
|
||||
completion protocol;
|
||||
- you need **multi-job fan-out** across tmux sessions with safe job claiming;
|
||||
- you want a clean PoC → authenticated-broker upgrade path.
|
||||
|
||||
**Do NOT use when:**
|
||||
- a one-shot `claude -p '…'` that returns inline is enough (no async signal
|
||||
needed) — just use the [claude-code](../claude-code/SKILL.md) skill directly;
|
||||
- you need request/response RPC or large artifact transfer (this is a
|
||||
one-direction event stream, not a data bus);
|
||||
- the payload would carry secrets and you're still on the public broker — move
|
||||
to the own-broker stage first.
|
||||
|
||||
## Quick Start
|
||||
|
||||
The one-line wrapper handles register + subscriber-first + agent launch. If
|
||||
you're new, **start here** and only fall back to the manual 5-step flow when
|
||||
you need finer control.
|
||||
The `multi-agent-mux-delegate-job` bash wrapper handles job registration, subscriber management, agent session targeting, and validation hooks:
|
||||
|
||||
```bash
|
||||
# 1) one line: register → start subscriber → launch agent in tmux
|
||||
# (uses public broker by default; last stdout line is the audit-log dir)
|
||||
# 1) Submit a new job to a targeted agent session (e.g. tmux session name 'demo')
|
||||
multi-agent-mux-delegate-job submit \
|
||||
--agent claude-code \
|
||||
--prompt "정렬 문제 10개를 만들어 sort_problems.md로 저장" \
|
||||
--workdir /path/to/project \
|
||||
--agent-session tmux:demo \
|
||||
--agent <claude-code|hermes-agent|agy-agent|cline-agent|human> \
|
||||
--agent-session tmux:<session_name> \
|
||||
--prompt "Task description or instructions here" \
|
||||
--timeout 3600 --idle-timeout 120
|
||||
# → stdout: registered job: <JID>
|
||||
# subscriber pid: …
|
||||
# agent launched in tmux session: demo
|
||||
# subscriber output: <one line per event>
|
||||
# /path/to/project/.mam/delegate_job_logs/<JID> ← audit log dir
|
||||
|
||||
# 2) at any time, query the job or its audit log
|
||||
multi-agent-mux-delegate-job status --job <JID>
|
||||
multi-agent-mux-delegate-job logs <JID> # pretty timeline
|
||||
multi-agent-mux-delegate-job logs --list # every job, live status
|
||||
# 2) Submit a job with a feedback loop (Worker-Reviewer Loop)
|
||||
multi-agent-mux-delegate-job submit \
|
||||
--agent <worker_agent> --agent-session tmux:<worker_session> \
|
||||
--type loop --reviewer <reviewer_agent> --reviewer-session tmux:<reviewer_session> \
|
||||
--prompt "Task description"
|
||||
|
||||
# 3) run a user-supplied validator against the job's artifacts
|
||||
multi-agent-mux-delegate-job verify --job <JID> --validate ./validate.sh
|
||||
# 3) Check job status and audit logs
|
||||
multi-agent-mux-delegate-job status --job <JOB_ID>
|
||||
multi-agent-mux-delegate-job logs <JOB_ID> # Chronological log of events
|
||||
multi-agent-mux-delegate-job list # Summary of all registered jobs
|
||||
|
||||
# 4) Verify job artifacts with a validation script
|
||||
multi-agent-mux-delegate-job verify --job <JOB_ID> --validate ./validate.sh
|
||||
```
|
||||
|
||||
The wrapper enforces the **subscribe-before-publish** ordering and **forwards
|
||||
the freshly-minted `JOB_ID` into the agent's prompt** (so the agent calls
|
||||
`publish_event.py --job <JID>` with the right id — see Pitfall §"Wrong job_id
|
||||
propagated to the agent"). When you need finer control, the manual flow is:
|
||||
---
|
||||
|
||||
```bash
|
||||
# Manual 5-step (same outcome, more knobs)
|
||||
PY=.venv/bin/python
|
||||
SKILL=./.agents/skills/multi-agent-mux-delegate-job/scripts
|
||||
## Task Delegation Types
|
||||
|
||||
# 1) register
|
||||
JID=$($PY "$SKILL/registry.py" register \
|
||||
--prompt "…" --agent claude-code --agent-session tmux:demo \
|
||||
--timeout 3600 --idle-timeout 120)
|
||||
Supported job types include:
|
||||
- `direct` (default): Single agent execution (direct tasking).
|
||||
- `loop` (Worker-Reviewer Loop): Alternates worker execution and reviewer evaluation until reviewer approves (`PASS`) or iterations run out.
|
||||
- `discuss` (Research & Discussion): Collaboration between two agents to reach a consensus (e.g., agreeing on a design or plan).
|
||||
|
||||
# 2) START THE SUBSCRIBER FIRST (MQTT does not queue non-retained msgs)
|
||||
$PY "$SKILL/job_subscriber.py" --job "$JID" --timeout 3600 --idle-timeout 120 &
|
||||
For detailed state machine diagrams and configurations, see [DELEGATION_TYPES.md](./DELEGATION_TYPES.md).
|
||||
|
||||
# 3) pass JID to the agent and instruct it to publish events with --job "$JID"
|
||||
# (don't hard-code a job id you saw earlier — see Pitfall §"Wrong job_id")
|
||||
---
|
||||
|
||||
# 4) on completion the subscriber prints events and exits 0/1/2
|
||||
## The Event Protocol Contract
|
||||
|
||||
# 5) inspect any time
|
||||
$PY "$SKILL/registry.py" get --job "$JID"
|
||||
$PY "$SKILL/registry.py" logs "$JID" # positional job id
|
||||
$PY "$SKILL/registry.py" logs --list
|
||||
```
|
||||
Every agent participating in the delegation contract must follow the same lifecycle publishing protocol using `publish_event.py`:
|
||||
|
||||
## Job Protocol
|
||||
1. **On Start**: Publish `started` event.
|
||||
`python3 .agents/skills/multi-agent-mux-delegate-job/scripts/publish_event.py --job "$JOB_ID" --event started`
|
||||
2. **On Tool/Permission Prompt**: Publish `permission_required` event.
|
||||
`python3 ... --job "$JOB_ID" --event permission_required --detail "<tool>:<reason>"`
|
||||
3. **On Progress Update (Optional)**: Publish `progress` event.
|
||||
`python3 ... --job "$JOB_ID" --event progress --detail "<status_update>"`
|
||||
4. **On Success**: Publish `completed` event.
|
||||
`python3 ... --job "$JOB_ID" --event completed --detail "<summary>"` (Reviewer should include `"PASS"` in the detail to approve).
|
||||
5. **On Failure/Feedback**: Publish `error` event.
|
||||
`python3 ... --job "$JOB_ID" --event error --detail "<reason_or_feedback>"`
|
||||
|
||||
One topic per job: `python/mqtt/jobs/<job_id>/events`. Payload (JSON, UTF-8,
|
||||
`schema_version=1`):
|
||||
|
||||
```json
|
||||
{ "schema_version": 1, "seq": 7, "job_id": "abc12345",
|
||||
"event": "started|permission_required|progress|completed|error",
|
||||
"timestamp": "2026-06-19T09:32:00Z", "detail": "generalised text",
|
||||
"data": { "optional": "metadata" } }
|
||||
```
|
||||
|
||||
- `seq` is monotonic per job (first = 1); the subscriber uses it to spot
|
||||
reorder/duplication.
|
||||
- `timestamp` is advisory — timeouts are measured from **receive** time.
|
||||
- `detail`/`data` carry **no** secrets or absolute paths.
|
||||
- A `schema_version` or `job_id` mismatch is **dropped** (defensive parsing).
|
||||
|
||||
`started` and `completed`/`error` are the mandatory bookends; `completed`→exit 0,
|
||||
`error`→exit 1. Full catalogue + production `auth_token` handling:
|
||||
[`job-protocol.md`](./job-protocol.md).
|
||||
|
||||
## Registry Format
|
||||
|
||||
```
|
||||
.mam/jobs/<id>.json # metadata record (single source of truth)
|
||||
.mam/jobs/<id>.events.log # append-only JSON-lines log (debug, optional)
|
||||
.mam/jobs/.lock # fcntl advisory lock for the registry
|
||||
```
|
||||
|
||||
The record holds `status`, `prompt`, `agent`, `agent_session`, a `broker` block,
|
||||
`topic_prefix`, `timeout_sec`/`idle_timeout_sec`, `expected_artifacts`,
|
||||
`last_seq`, and (production) `auth_token`. Because the `broker` block lives in
|
||||
the record, `publish_event.py` connects from the registry alone. Concurrency,
|
||||
the atomic rename trick, and multi-session job claiming are in
|
||||
[`registry.md`](./registry.md).
|
||||
---
|
||||
|
||||
## Audit Logs
|
||||
|
||||
Every job's lifecycle is mirrored to a **persistent, append-only audit log**
|
||||
under `.mam/delegate_job_logs/` (override with `DELEGATE_JOB_LOGS_DIR`;
|
||||
default `<cwd>/.mam/delegate_job_logs`). Unlike the registry — live state
|
||||
mutated in place and liable to be cleaned up — the audit log is durable
|
||||
history you can replay after the fact. It is git-ignored.
|
||||
Job lifecycle execution events are persistently mirrored to an append-only log under `.mam/delegate_job_logs/<job_id>/` (containing `meta.json`, `events.ndjson`, and `status.json`). Use `multi-agent-mux-delegate-job logs <job_id>` to view the timeline.
|
||||
|
||||
```
|
||||
.mam/delegate_job_logs/<job_id>/
|
||||
meta.json # registration snapshot: prompt, agent, broker, timeouts, …
|
||||
events.ndjson # append-only, one JSON event per line, in time order
|
||||
status.json # current status only (fast point-query)
|
||||
```
|
||||
---
|
||||
|
||||
**What is logged, automatically:**
|
||||
## Best Practices and Pitfalls
|
||||
|
||||
| When | `events.ndjson` line | Written by |
|
||||
|------|----------------------|------------|
|
||||
| job registered | `registered` (also seeds meta.json + status.json) | `registry.register_job` |
|
||||
| any status change | `status_changed` (`from`/`to`; also rewrites status.json) | `update_job_status`, `pick_pending` |
|
||||
| event published | `published` (carries the exact payload — reproducible) | `publish_event.py` |
|
||||
| event received | `received` (subscriber's external view) | `job_subscriber.py` |
|
||||
|
||||
Both the emitter side (`published`) and the observer side (`received`) are
|
||||
recorded, so a dropped publish or a missed receive is still visible from the
|
||||
other. Every write is **best-effort and isolated** — an fcntl-locked append
|
||||
guarded by `try/except` that only ever emits a `logger.warning`, so a logging
|
||||
failure can never break a publish, a subscribe, or a registry write. stdout is
|
||||
never touched.
|
||||
|
||||
**Reading them:**
|
||||
|
||||
```bash
|
||||
multi-agent-mux-delegate-job logs <job_id> # pretty-print one job's timeline
|
||||
multi-agent-mux-delegate-job logs --list # summarise every logged job (with live status)
|
||||
# or directly via the registry CLI:
|
||||
$PY scripts/registry.py logs <job_id> [--tail N] [--json]
|
||||
$PY scripts/registry.py logs --list [--json]
|
||||
```
|
||||
|
||||
`submit` prints the job's audit-log directory as its last stdout line, so a
|
||||
caller can `tail -n1` to locate it.
|
||||
|
||||
## Broker Setup
|
||||
|
||||
| Stage | Broker | Auth | Transport |
|
||||
|-------|--------|------|-----------|
|
||||
| PoC | `broker.hivemq.com` | none | 1883 plaintext |
|
||||
| Production | self-hosted Mosquitto/EMQX | user/pass + ACL | 8883 TLS |
|
||||
|
||||
All connection settings come from env (`MQTT_BROKER`, `MQTT_PORT`, `MQTT_TLS`,
|
||||
`MQTT_USERNAME`/`MQTT_PASSWORD`, `MQTT_CA_CERTS`, …) resolved by
|
||||
`broker_config_from_env()`, with the registry `broker.*` block overriding per
|
||||
job. Moving to your own broker is **config only**: install Mosquitto, set
|
||||
`persistence true` + `acl_file` + `password_file` + a TLS `listener 8883`, grant
|
||||
the worker `write python/mqtt/jobs/+/events` and Hermes `read`, then flip
|
||||
`MQTT_TLS=1` and fill the registry `broker.*`. Step-by-step (conf, ACL,
|
||||
`mosquitto_passwd`, self-signed/private-CA certs, cut-over verification):
|
||||
[`mqtt-broker-setup.md`](./mqtt-broker-setup.md).
|
||||
|
||||
## Agent Adapters
|
||||
|
||||
Each agent voluntarily follows the contract: receive a `JOB_ID` (or registry
|
||||
path), call `publish_event.py` at lifecycle points, exit 0/1/2. **The contract
|
||||
in one line**: every event call uses `--job "$JOB_ID"` where `$JOB_ID` is the
|
||||
**freshly-issued id from the registry record for *this* delegation** — never a
|
||||
job_id you saw in an earlier session (Pitfall §"Wrong job_id propagated to the
|
||||
agent").
|
||||
|
||||
- **claude-code** — Claude Code calls `publish_event.py` via its Bash tool at
|
||||
lifecycle points. `submit --mode tmux` injects a prompt that already names
|
||||
`$JOB_ID`; if you drive claude manually, hand it the id explicitly. Reference
|
||||
instruction block (the wrapper injects something equivalent):
|
||||
|
||||
```text
|
||||
Your job_id is "$JOB_ID" (read it from the registry record for this delegation —
|
||||
do not reuse any job_id you saw before).
|
||||
|
||||
On start: $PY multi-agent-mux-delegate-job/scripts/publish_event.py --job "$JOB_ID" --event started
|
||||
On permission: $PY … --job "$JOB_ID" --event permission_required --detail "<tool>:<what>"
|
||||
On progress: $PY … --job "$JOB_ID" --event progress --detail "<short status>"
|
||||
On success: $PY … --job "$JOB_ID" --event completed --detail "<one-line summary>"
|
||||
On failure: $PY … --job "$JOB_ID" --event error --detail "<one-line reason>"
|
||||
|
||||
Task: <the user's prompt>
|
||||
|
||||
The subscriber for "$JOB_ID" is already running; your completed/error event
|
||||
ends the job. Exit codes: 0 completed, 1 error, 2 publish failure.
|
||||
```
|
||||
|
||||
See [claude-code](../claude-code/SKILL.md) for tmux orchestration patterns.
|
||||
- **codex** — same contract. Invoke `codex exec "<instruction-block-above>"` or
|
||||
wire `publish_event.py` as an MCP tool so the agent can call it directly.
|
||||
- **opencode** — wire `publish_event.py` as a tool/command the agent can call;
|
||||
identical event points.
|
||||
- **human** — a person does the work, reads the registry record, then runs
|
||||
`publish_event.py --job <id> --event completed` (or `error`) by hand.
|
||||
|
||||
## User Interface
|
||||
|
||||
The [`multi-agent-mux-delegate-job`](./multi-agent-mux-delegate-job) bash wrapper bundles register +
|
||||
subscribe-first + run-agent + validate:
|
||||
|
||||
```bash
|
||||
multi-agent-mux-delegate-job submit --agent claude-code \
|
||||
--prompt "정렬 문제 10개를 만들어 sort_problems.md로 저장" \
|
||||
--workdir /path/to/project --timeout 3600 [--validate ./validate.sh]
|
||||
multi-agent-mux-delegate-job status --job <id> # one record, pretty-printed
|
||||
multi-agent-mux-delegate-job list # all jobs, one line each
|
||||
multi-agent-mux-delegate-job verify --job <id> --validate ./validate.sh # runs it, reports exit code
|
||||
multi-agent-mux-delegate-job wait [--job <id>] # block until terminal (else --wait-any)
|
||||
```
|
||||
|
||||
`submit` **always starts the subscriber before the agent** (the ordering
|
||||
dependency), runs the agent in `--mode print` (one-shot) or `--mode tmux`, and
|
||||
calls `--validate` afterward if given. The skill automates job-id generation,
|
||||
registry creation, broker resolution, subscriber-first ordering, agent launch,
|
||||
and completion detection; it does **not** automate the agent's internals or your
|
||||
business-logic validation — those are hooks you fill (`validate.sh` reads
|
||||
`$JOB_ID`/`$REGISTRY_DIR`).
|
||||
|
||||
## Common Pitfalls
|
||||
|
||||
- **Publishing before subscribing** — MQTT does not queue non-retained messages
|
||||
for absent subscribers. Start `job_subscriber.py` *before* the agent, or rely
|
||||
on retained terminal events (production). `submit` enforces this.
|
||||
- **Wrong job_id propagated to the agent** — the wrapper prints a fresh `JOB_ID`
|
||||
on every `submit`. If your agent instruction (or the wrapper's prompt template)
|
||||
hard-codes an old job_id, the agent calls `publish_event.py --job <wrong>`,
|
||||
the subscriber's defensive parser drops it as a `job_id` mismatch, and the
|
||||
delegator waits until idle timeout (exit 2). Fix: instruct the agent to
|
||||
**read the job_id from the registry record for *this* delegation** (or pass it
|
||||
in via env / `--prompt` interpolation), never from prior runs. `submit`'s
|
||||
default prompt template interpolates `$JOB_ID` for you — if you build a custom
|
||||
prompt, do the same.
|
||||
- **tmux session name collision** — `submit --mode tmux` derives the session
|
||||
name from `--agent-session tmux:<name>` (default `tmux:claude`). If a session
|
||||
with that name is already attached (e.g. you ran the demo and the previous
|
||||
session is still open), `tmux new-session -d -s <name>` fails and the agent
|
||||
never launches. Pick a unique `--agent-session` per concurrent delegation
|
||||
(e.g. `tmux:demo`, `tmux:claude-a`, `tmux:claude-b`) or kill the stale one
|
||||
(`tmux kill-session -t claude`) before re-running.
|
||||
- **Timeout before `started`** — a cold-starting agent may not emit `started`
|
||||
for a while; the wall-clock timeout starts at subscribe time so a stuck agent
|
||||
still terminates. Don't set `--timeout` so low you false-positive a slow start.
|
||||
- **No retry on publish** — a dropped `completed` would hang the delegator
|
||||
forever; `publish_event.py` retries with exponential backoff and exits 2 if it
|
||||
still fails, so the delegator is never left waiting silently.
|
||||
- **QoS-1 duplicates / reorders** — a terminal event can arrive twice, or
|
||||
`error` can trail `completed`; the subscriber's terminal state machine
|
||||
finalises each job once and ignores the rest.
|
||||
- **Trusting the public broker** — anyone can publish there; never make a real
|
||||
decision on a PoC signal. Add `auth_token` + an authenticated broker first.
|
||||
- **Secrets in `detail`/`data`** — keep payloads generalised; no paths, keys, or
|
||||
tokens (except the production `auth_token` in `data`).
|
||||
|
||||
## Subagent Orchestration Pattern
|
||||
|
||||
When using this skill from a Hermes `delegate_task` subagent to dispatch work to
|
||||
a coding-agent CLI (agy/claude) running in a tmux session, the following pattern
|
||||
has been verified (2026-06-21, 6-batch refactoring sprint):
|
||||
|
||||
### Roles
|
||||
- **Main worker** (implementation): one agent session (e.g. `agy-new`) receives
|
||||
brief files and executes code changes.
|
||||
- **Reviewers** (spec compliance + code quality): two other agent sessions
|
||||
(e.g. `agy-existing`, `claude-existing`) review the diff in parallel.
|
||||
- **Hermes** (orchestrator): dispatches subagents, verifies diffs, commits,
|
||||
and falls back to direct fixes when reviewers find issues.
|
||||
|
||||
### Key lessons learned
|
||||
1. **Brief delivery via file path** — don't paste long briefs inline via
|
||||
`tmux send-keys`; the TUI may swallow them. Instead, send a short instruction
|
||||
like "follow /tmp/batch1-brief.md" and let the agent read the file.
|
||||
2. **Polling vs MQTT subscriber** — for short tasks (<5min), pane polling
|
||||
(`capture-pane` + grep for completion markers) is simpler and more reliable
|
||||
than registering a job via `registry.py` + `job_subscriber.py`. Use MQTT
|
||||
subscriber only for long-running jobs (>5min) where push notification matters.
|
||||
3. **Reviewers catch different bugs** — in practice, agy (Flash) caught
|
||||
semantic issues (slash matching, export scope), while claude (Opus) caught
|
||||
API signature mismatches (paho v2 5-arg vs 4-arg `on_disconnect`). Two
|
||||
reviewers with different models provide complementary coverage.
|
||||
4. **Hermes fallback fix** — when reviewers find a small, well-defined issue
|
||||
(wrong argument count, missing slash), Hermes should fix it directly rather
|
||||
than re-dispatching the implementer. This saves a full round-trip.
|
||||
5. **Batch grouping** — group 2-3 FW items per batch when they touch different
|
||||
files (no file overlap). This amortises the dispatch overhead. Items touching
|
||||
the same file must be in separate batches to avoid conflicts.
|
||||
6. **Pane Snapshots & Truncation Prevention** — to prevent long agent responses from being scrolled out and truncated due to TUI viewport limitations, enforce the following snapshotting pattern:
|
||||
- Immediately after dispatching a brief, capture the pre-brief pane buffer via `capture-pane -S -200`.
|
||||
- During long execution, run a background loop taking incremental snapshots (e.g. every 30 seconds `>> /tmp/pane-snap.txt`).
|
||||
- Immediately after job termination, capture the entire final pane state to ensure no terminal logs are lost.
|
||||
|
||||
## Verification Checklist
|
||||
|
||||
- [ ] `started` → `completed` over the public broker: subscriber prints the
|
||||
lines and exits **0**.
|
||||
- [ ] `error` path: subscriber exits **1**.
|
||||
- [ ] timeout path: no terminal event within `--timeout`/`--idle-timeout` →
|
||||
exit **2**.
|
||||
- [ ] polluted payload (bad JSON, wrong `schema_version`, wrong `job_id`) is
|
||||
dropped with a warning, not crashed on.
|
||||
- [ ] one tmux session processes two registry jobs in sequence; a second
|
||||
session with a different `agent_session` claims only its own.
|
||||
- [ ] broker cut-over: same scripts reach an authenticated TLS broker with env
|
||||
changes only; a credential without write ACL is rejected; a late
|
||||
subscriber still receives the retained terminal event.
|
||||
- [ ] `publisher.py`/`subscriber.py`/`README.md` demo on `python/mqtt/sample`
|
||||
still works unchanged (regression).
|
||||
- [ ] **audit log integrity** — for a completed job,
|
||||
`.mam/delegate_job_logs/<JID>/events.ndjson` contains `registered` →
|
||||
`received started` → `published completed` (in that order), and
|
||||
`status.json.status == "completed"` matches the registry record. A
|
||||
logging failure (e.g. read-only log dir) does not break the publish or
|
||||
subscribe path — only a `logger.warning` is emitted.
|
||||
- [ ] **end-to-end demo smoke** — run
|
||||
`multi-agent-mux-delegate-job submit --agent claude-code --agent-session tmux:demo-smoke
|
||||
--prompt "echo hello and call publish_event.py --job <JID>
|
||||
--event completed" --timeout 120` and confirm
|
||||
(a) registered job id echoed, (b) subscriber pid echoed, (c) tmux session
|
||||
name printed, (d) `events.ndjson` grows as the agent runs, (e) final
|
||||
stdout line is the audit-log dir.
|
||||
- **Subscribe-Before-Publish**: The subscriber must be running before the agent starts publishing. The `submit` command handles this automatically by launching the subscriber in the background first.
|
||||
- **Fresh job_id Propagation**: Make sure the worker agent receives the correct `JOB_ID` generated for the current run, rather than reusing stale IDs from previous sessions.
|
||||
- **Brief delivery via file path**: For long or complex prompts, write the instructions to a file (e.g. `/tmp/task-brief.md`) and pass a short prompt pointing to the file path to prevent terminal buffer overflows.
|
||||
- **Batch Grouping**: Group non-overlapping tasks into batches to parallelize execution across multiple agent sessions, reducing overhead.
|
||||
|
||||
@@ -16,6 +16,13 @@ set -euo pipefail
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
|
||||
# Load local .env if it exists in current dir or workspace root
|
||||
if [[ -f .env ]]; then
|
||||
set -a; source .env; set +a
|
||||
elif [[ -f "$SCRIPT_DIR/../../.env" ]]; then
|
||||
set -a; source "$SCRIPT_DIR/../../.env"; set +a
|
||||
fi
|
||||
|
||||
# Pick an interpreter: prefer a project .venv, else python3.
|
||||
pick_python() {
|
||||
local py_bin
|
||||
@@ -46,6 +53,8 @@ multi-agent-mux-delegate-job <command> [options]
|
||||
submit --agent <name> --prompt <text> [--workdir <dir>] [--agent-session <label>]
|
||||
[--timeout <sec>] [--idle-timeout <sec>] [--validate <script>]
|
||||
[--registry-dir <dir>] [--dry-run]
|
||||
[--type <direct|loop|discuss>] [--reviewer <reviewer_agent>]
|
||||
[--reviewer-session <reviewer_session>] [--max-iterations <count>]
|
||||
# The skill is tmux-interactive only; --mode print was removed.
|
||||
status --job <id> [--registry-dir <dir>]
|
||||
list [--registry-dir <dir>]
|
||||
@@ -59,6 +68,7 @@ EOF
|
||||
AGENT="claude-code"; PROMPT=""; WORKDIR="$(pwd)"; AGENT_SESSION="tmux:claude"
|
||||
TIMEOUT=3600; IDLE_TIMEOUT=120; VALIDATE=""; DRY_RUN=0
|
||||
JOB_ID=""; REGISTRY_DIR="$REGISTRY_DIR_DEFAULT"
|
||||
TYPE="direct"; REVIEWER="hermes"; REVIEWER_SESSION="tmux:hermes"; MAX_ITERATIONS=5
|
||||
|
||||
parse_opts() {
|
||||
while [[ $# -gt 0 ]]; do
|
||||
@@ -73,6 +83,10 @@ parse_opts() {
|
||||
--job) JOB_ID="$2"; shift 2;;
|
||||
--registry-dir) REGISTRY_DIR="$2"; shift 2;;
|
||||
--dry-run) DRY_RUN=1; shift;;
|
||||
--type) TYPE="$2"; shift 2;;
|
||||
--reviewer) REVIEWER="$2"; shift 2;;
|
||||
--reviewer-session) REVIEWER_SESSION="$2"; shift 2;;
|
||||
--max-iterations) MAX_ITERATIONS="$2"; shift 2;;
|
||||
*) echo "unknown option: $1" >&2; usage; exit 1;;
|
||||
esac
|
||||
done
|
||||
@@ -88,26 +102,29 @@ cmd_submit() {
|
||||
# 1) register job (prints the new job id)
|
||||
JOB_ID="$("$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" register \
|
||||
--prompt "$PROMPT" --agent "$AGENT" --agent-session "$AGENT_SESSION" \
|
||||
--timeout "$TIMEOUT" --idle-timeout "$IDLE_TIMEOUT")"
|
||||
--timeout "$TIMEOUT" --idle-timeout "$IDLE_TIMEOUT" \
|
||||
--job-type "$TYPE" --reviewer "$REVIEWER" --reviewer-session "$REVIEWER_SESSION" \
|
||||
--max-iterations "$MAX_ITERATIONS")"
|
||||
echo "registered job: $JOB_ID"
|
||||
|
||||
# 2) START THE SUBSCRIBER FIRST (ordering dependency — MQTT does not queue
|
||||
# non-retained messages for absent subscribers).
|
||||
local logf="$REGISTRY_DIR/$JOB_ID.subscriber.out"
|
||||
"$PY" "$SCRIPT_DIR/scripts/job_subscriber.py" --registry-dir "$REGISTRY_DIR" \
|
||||
--job "$JOB_ID" --timeout "$TIMEOUT" --idle-timeout "$IDLE_TIMEOUT" \
|
||||
>"$logf" 2>&1 &
|
||||
local sub_pid=$!
|
||||
echo "subscriber pid: $sub_pid (log: $logf)"
|
||||
sleep 1 # give the subscriber time to CONNACK + SUBSCRIBE before the agent runs
|
||||
if [[ "$TYPE" == "direct" ]]; then
|
||||
# 2) START THE SUBSCRIBER FIRST (ordering dependency — MQTT does not queue
|
||||
# non-retained messages for absent subscribers).
|
||||
local logf="$REGISTRY_DIR/$JOB_ID.subscriber.out"
|
||||
"$PY" "$SCRIPT_DIR/scripts/job_subscriber.py" --registry-dir "$REGISTRY_DIR" \
|
||||
--job "$JOB_ID" --timeout "$TIMEOUT" --idle-timeout "$IDLE_TIMEOUT" \
|
||||
>"$logf" 2>&1 &
|
||||
local sub_pid=$!
|
||||
echo "subscriber pid: $sub_pid (log: $logf)"
|
||||
sleep 1 # give the subscriber time to CONNACK + SUBSCRIBE before the agent runs
|
||||
|
||||
# 3) run the agent (or print the command for dry-run / missing binary)
|
||||
local pub="$PY $SCRIPT_DIR/scripts/publish_event.py --registry-dir $REGISTRY_DIR --job $JOB_ID"
|
||||
# NOTE: the agent MUST use --job "$JOB_ID" (the one we just minted). Hard-coding
|
||||
# an id from an earlier session is the #1 reason a delegated job sits idle and
|
||||
# times out (see SKILL.md "Wrong job_id propagated to the agent"). We make the
|
||||
# freshness explicit in the instruction header.
|
||||
local instructions="Your job_id is \"$JOB_ID\" (the one just registered for THIS delegation — read it from the registry record, do NOT reuse any job_id you saw in earlier runs).
|
||||
# 3) run the agent (or print the command for dry-run / missing binary)
|
||||
local pub="$PY $SCRIPT_DIR/scripts/publish_event.py --registry-dir $REGISTRY_DIR --job $JOB_ID"
|
||||
# NOTE: the agent MUST use --job "$JOB_ID" (the one we just minted). Hard-coding
|
||||
# an id from an earlier session is the #1 reason a delegated job sits idle and
|
||||
# times out (see SKILL.md "Wrong job_id propagated to the agent"). We make the
|
||||
# freshness explicit in the instruction header.
|
||||
local instructions="Your job_id is \"$JOB_ID\" (the one just registered for THIS delegation — read it from the registry record, do NOT reuse any job_id you saw in earlier runs).
|
||||
|
||||
On start run: $pub --event started.
|
||||
On permission/tool prompt run: $pub --event permission_required --detail '<tool>:<what>'.
|
||||
@@ -119,40 +136,185 @@ The subscriber for this job_id is already running; your completed/error event en
|
||||
|
||||
Task: $PROMPT"
|
||||
|
||||
run_agent "$JOB_ID" "$instructions"
|
||||
run_agent "$JOB_ID" "$instructions"
|
||||
|
||||
# 4) optional validation hook
|
||||
if [[ -n "$VALIDATE" ]]; then
|
||||
echo "running validation: $VALIDATE"
|
||||
if JOB_ID="$JOB_ID" REGISTRY_DIR="$REGISTRY_DIR" bash "$VALIDATE"; then
|
||||
echo "validation: PASS"
|
||||
else
|
||||
local rc=$?
|
||||
echo "validation: FAIL (exit $rc)"
|
||||
# 4) optional validation hook
|
||||
if [[ -n "$VALIDATE" ]]; then
|
||||
echo "running validation: $VALIDATE"
|
||||
if JOB_ID="$JOB_ID" REGISTRY_DIR="$REGISTRY_DIR" bash "$VALIDATE"; then
|
||||
echo "validation: PASS"
|
||||
else
|
||||
local rc=$?
|
||||
echo "validation: FAIL (exit $rc)"
|
||||
fi
|
||||
fi
|
||||
|
||||
if [[ "$DRY_RUN" == "1" ]]; then
|
||||
# In dry-run we never started a real subscriber (the wrapper short-circuits
|
||||
# before launching one), but the wait below would still try to join the
|
||||
# background sub_pid from cmd_submit. Skip both the wait and the subscriber
|
||||
# log dump; the user just wants to see the instruction that would have run.
|
||||
local logs_root_dry="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
|
||||
echo "$logs_root_dry/$JOB_ID"
|
||||
return 0
|
||||
fi
|
||||
|
||||
wait "$sub_pid" || true
|
||||
echo "subscriber output:"; cat "$logf" || true
|
||||
|
||||
# Last stdout line: the persistent audit-log dir for this job (see SKILL.md
|
||||
# "Audit Logs"). Callers can scrape `tail -n1` to find it.
|
||||
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
|
||||
echo "$logs_root/$JOB_ID"
|
||||
else
|
||||
# Implement loop/discuss orchestrator
|
||||
local iteration=1
|
||||
local current_prompt="$PROMPT"
|
||||
local current_session="$AGENT_SESSION"
|
||||
local current_role="worker"
|
||||
|
||||
if [[ "$DRY_RUN" == "1" ]]; then
|
||||
echo "[dry-run] orchestrator loop would start for job: $JOB_ID type: $TYPE"
|
||||
echo "worker session: $AGENT_SESSION, reviewer session: $REVIEWER_SESSION"
|
||||
local logs_root_dry="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
|
||||
echo "$logs_root_dry/$JOB_ID"
|
||||
return 0
|
||||
fi
|
||||
|
||||
while true; do
|
||||
echo "=================================================="
|
||||
echo "Iteration $iteration - Role: $current_role"
|
||||
echo "Session: $current_session"
|
||||
echo "=================================================="
|
||||
|
||||
# Update job details in registry
|
||||
"$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" update \
|
||||
--job "$JOB_ID" \
|
||||
--agent-session "$current_session" \
|
||||
--prompt "$current_prompt" \
|
||||
--iteration "$iteration" \
|
||||
--status "pending"
|
||||
|
||||
# Start subscriber
|
||||
local logf="$REGISTRY_DIR/${JOB_ID}.iter_${iteration}_${current_role}.subscriber.out"
|
||||
"$PY" "$SCRIPT_DIR/scripts/job_subscriber.py" --registry-dir "$REGISTRY_DIR" \
|
||||
--job "$JOB_ID" --timeout "$TIMEOUT" --idle-timeout "$IDLE_TIMEOUT" \
|
||||
>"$logf" 2>&1 &
|
||||
local sub_pid=$!
|
||||
echo "subscriber pid: $sub_pid (log: $logf)"
|
||||
sleep 1
|
||||
|
||||
# Format instruction block
|
||||
local pub="$PY $SCRIPT_DIR/scripts/publish_event.py --registry-dir $REGISTRY_DIR --job $JOB_ID"
|
||||
local instructions="Your job_id is \"$JOB_ID\" (the one just registered for THIS delegation — read it from the registry record, do NOT reuse any job_id you saw in earlier runs).
|
||||
|
||||
On start run: $pub --event started.
|
||||
On permission/tool prompt run: $pub --event permission_required --detail '<tool>:<what>'.
|
||||
On progress (optional): $pub --event progress --detail '<short status>'.
|
||||
On success run: $pub --event completed --detail '<one-line summary>'.
|
||||
On failure run: $pub --event error --detail '<one-line reason>'.
|
||||
|
||||
The subscriber for this job_id is already running; your completed/error event ends the job. Exit codes: 0 completed, 1 error, 2 publish failure.
|
||||
|
||||
Task: $current_prompt"
|
||||
|
||||
# Trigger agent
|
||||
run_agent "$JOB_ID" "$instructions" "$current_session"
|
||||
|
||||
# Wait for subscriber
|
||||
local sub_rc=0
|
||||
wait "$sub_pid" || sub_rc=$?
|
||||
echo "subscriber output:"; cat "$logf" || true
|
||||
|
||||
# Check job status based on subscriber exit code
|
||||
local job_status="running"
|
||||
if [[ $sub_rc -eq 0 ]]; then
|
||||
job_status="completed"
|
||||
elif [[ $sub_rc -eq 1 ]]; then
|
||||
job_status="error"
|
||||
else
|
||||
job_status="timeout"
|
||||
fi
|
||||
|
||||
echo "Job role $current_role finished with status: $job_status"
|
||||
|
||||
# Retrieve feedback from the last event
|
||||
local feedback
|
||||
feedback="$("$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" get-feedback --job "$JOB_ID")"
|
||||
echo "Feedback/Detail: $feedback"
|
||||
|
||||
if [[ "$current_role" == "worker" ]]; then
|
||||
if [[ "$job_status" != "completed" ]]; then
|
||||
echo "Worker did not complete successfully (status: $job_status). Terminating workflow."
|
||||
break
|
||||
fi
|
||||
|
||||
# Worker completed successfully, now switch to reviewer
|
||||
current_role="reviewer"
|
||||
current_session="$REVIEWER_SESSION"
|
||||
# Build reviewer prompt based on type
|
||||
if [[ "$TYPE" == "loop" ]]; then
|
||||
current_prompt="Review the changes/artifacts generated for job $JOB_ID. Check if they meet the requirements. If correct, publish completed event with 'PASS'. If there are issues, publish error event with detailed feedback/nits. CRITICAL: When raising issues or giving a review, you MUST include the exact reason for the issue and a clear direction for improvement (문제 제시에 대한 이유와 확실한 개선 방향을 반드시 포함해야 합니다)."
|
||||
elif [[ "$TYPE" == "discuss" ]]; then
|
||||
current_prompt="Read draft/documents generated for job $JOB_ID. Review the feasibility and content. Write your feedback/objections. If you agree with the plan, reply with 'AGREE'."
|
||||
fi
|
||||
else
|
||||
if [[ "$job_status" != "completed" ]]; then
|
||||
echo "Reviewer did not complete successfully (status: $job_status). Terminating workflow."
|
||||
break
|
||||
fi
|
||||
|
||||
# Reviewer finished. Check if pass/agree
|
||||
local success=0
|
||||
if [[ "$TYPE" == "loop" ]]; then
|
||||
if [[ "${feedback,,}" == *"pass"* ]]; then
|
||||
success=1
|
||||
fi
|
||||
elif [[ "$TYPE" == "discuss" ]]; then
|
||||
if [[ "${feedback,,}" == *"agree"* ]]; then
|
||||
success=1
|
||||
fi
|
||||
fi
|
||||
|
||||
if [[ "$success" == "1" ]]; then
|
||||
echo "Reviewer approved the work. Finalizing job as completed."
|
||||
"$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" status --job "$JOB_ID" --set "completed"
|
||||
break
|
||||
else
|
||||
# Reviewer rejected/provided feedback. Increment & check max iterations
|
||||
if [[ $iteration -ge $MAX_ITERATIONS ]]; then
|
||||
echo "Max iterations ($MAX_ITERATIONS) reached without approval. Terminating workflow."
|
||||
"$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" status --job "$JOB_ID" --set "error"
|
||||
break
|
||||
fi
|
||||
|
||||
iteration=$((iteration + 1))
|
||||
current_role="worker"
|
||||
current_session="$AGENT_SESSION"
|
||||
current_prompt="The reviewer provided the following feedback for job $JOB_ID: $feedback. Please modify the code/artifacts to address these comments. CRITICAL: As the Developer Team Leader, you must thoroughly review the suggested modifications, verify their validity, adopt/implement them if valid, and if you judge any recommendation to be invalid, do NOT implement it but instead explain your reasons clearly in your response and send it back to the reviewer (수정안을 최대한 꼼꼼히 검토하여 타당성을 검증하고, 타당하다면 수렴하여 수정을 진행하되, 타당하지 않다고 판단되는 부분이 있다면 그 이유를 명확히 밝혀 리뷰어에게 전달하십시오)."
|
||||
fi
|
||||
fi
|
||||
done
|
||||
|
||||
# 4) optional validation hook
|
||||
if [[ -n "$VALIDATE" ]]; then
|
||||
echo "running validation: $VALIDATE"
|
||||
if JOB_ID="$JOB_ID" REGISTRY_DIR="$REGISTRY_DIR" bash "$VALIDATE"; then
|
||||
echo "validation: PASS"
|
||||
else
|
||||
local rc=$?
|
||||
echo "validation: FAIL (exit $rc)"
|
||||
fi
|
||||
fi
|
||||
|
||||
# Last stdout line: the persistent audit-log dir
|
||||
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
|
||||
echo "$logs_root/$JOB_ID"
|
||||
fi
|
||||
|
||||
if [[ "$DRY_RUN" == "1" ]]; then
|
||||
# In dry-run we never started a real subscriber (the wrapper short-circuits
|
||||
# before launching one), but the wait below would still try to join the
|
||||
# background sub_pid from cmd_submit. Skip both the wait and the subscriber
|
||||
# log dump; the user just wants to see the instruction that would have run.
|
||||
local logs_root_dry="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
|
||||
echo "$logs_root_dry/$JOB_ID"
|
||||
return 0
|
||||
fi
|
||||
|
||||
wait "$sub_pid" || true
|
||||
echo "subscriber output:"; cat "$logf" || true
|
||||
|
||||
# Last stdout line: the persistent audit-log dir for this job (see SKILL.md
|
||||
# "Audit Logs"). Callers can scrape `tail -n1` to find it.
|
||||
local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
|
||||
echo "$logs_root/$JOB_ID"
|
||||
}
|
||||
|
||||
run_agent() {
|
||||
local job_id="$1"; local instructions="$2"
|
||||
local job_id="$1"; local instructions="$2"; local target_session="${3:-$AGENT_SESSION}"
|
||||
# The skill is INTERACTIVE-ONLY. We never invoke `claude -p` or any other
|
||||
# one-shot print mode, because:
|
||||
# - claude -p exits the moment stdin is drained, so there's nothing to
|
||||
@@ -168,7 +330,7 @@ run_agent() {
|
||||
echo "[human agent] complete the task, then run publish_event.py --event completed"
|
||||
return
|
||||
fi
|
||||
local sess="${AGENT_SESSION#tmux:}"
|
||||
local sess="${target_session#tmux:}"
|
||||
|
||||
if [[ "$DRY_RUN" == "1" ]]; then
|
||||
echo "[dry-run] would delegate task to running agent '$AGENT' in tmux session '$sess' with instructions:"
|
||||
@@ -202,6 +364,7 @@ run_agent() {
|
||||
echo "살아있는 에이전트 세션 '$sess'에 작업을 위임합니다..."
|
||||
$_tmux set-buffer -b "job_buf_$job_id" "$instructions"
|
||||
$_tmux paste-buffer -b "job_buf_$job_id" -t "$sess"
|
||||
sleep 0.5
|
||||
$_tmux send-keys -t "$sess" C-m
|
||||
$_tmux delete-buffer -b "job_buf_$job_id"
|
||||
|
||||
|
||||
@@ -59,11 +59,11 @@ def _format_line(topic: str, payload: Dict[str, Any]) -> str:
|
||||
class _Watcher:
|
||||
"""Holds the shared queue + the set of job_ids we accept events for."""
|
||||
|
||||
def __init__(self, expected_job_ids: Set[str], expected_tokens: Dict[str, Optional[str]]):
|
||||
def __init__(self, expected_job_ids: Set[str], expected_tokens: Dict[str, Optional[str]], expected_seqs: Dict[str, int]):
|
||||
self.events: "queue.Queue[Tuple[str, Dict[str, Any]]]" = queue.Queue()
|
||||
self.expected = set(expected_job_ids)
|
||||
self.tokens = expected_tokens # job_id -> expected auth_token (or None)
|
||||
self.last_seq: Dict[str, int] = {jid: 0 for jid in expected_job_ids}
|
||||
self.last_seq = dict(expected_seqs)
|
||||
|
||||
def on_message(self, _client, _userdata, msg) -> None:
|
||||
# --- defensive parsing -------------------------------------------
|
||||
@@ -153,7 +153,8 @@ def main(argv=None) -> int:
|
||||
|
||||
expected_ids: Set[str] = {j["job_id"] for j in jobs}
|
||||
tokens = {j["job_id"]: j.get("auth_token") for j in jobs}
|
||||
watcher = _Watcher(expected_ids, tokens)
|
||||
seqs = {j["job_id"]: int(j.get("last_seq", 0)) for j in jobs}
|
||||
watcher = _Watcher(expected_ids, tokens, seqs)
|
||||
|
||||
# Resolve timeouts from CLI, falling back to the (first) job's settings.
|
||||
base_job = jobs[0]
|
||||
|
||||
@@ -59,6 +59,10 @@ def register_job(
|
||||
expected_artifacts: Optional[List[str]] = None,
|
||||
bits: int = 32,
|
||||
auth_token: Optional[str] = None,
|
||||
job_type: str = "direct",
|
||||
reviewer: Optional[str] = None,
|
||||
reviewer_session: Optional[str] = None,
|
||||
max_iterations: int = 5,
|
||||
) -> str:
|
||||
"""Create a new ``pending`` job record and return its id.
|
||||
|
||||
@@ -90,6 +94,11 @@ def register_job(
|
||||
"expected_artifacts": expected_artifacts or [],
|
||||
"last_seq": 0,
|
||||
"auth_token": auth_token,
|
||||
"job_type": job_type,
|
||||
"reviewer": reviewer,
|
||||
"reviewer_session": reviewer_session,
|
||||
"max_iterations": int(max_iterations),
|
||||
"iteration": 1,
|
||||
}
|
||||
with registry_lock(registry_dir):
|
||||
if mqtt_common._job_path(job_id, registry_dir).exists():
|
||||
@@ -164,7 +173,7 @@ def append_event(job_id: str, registry_dir: str, payload: Dict[str, Any]) -> Non
|
||||
# convenience re-export so callers can `from registry import load_job`
|
||||
__all__ = [
|
||||
"register_job", "pick_pending", "update_status", "load_job",
|
||||
"list_jobs", "append_event", "generate_job_id",
|
||||
"list_jobs", "append_event", "generate_job_id", "get_feedback",
|
||||
]
|
||||
|
||||
|
||||
@@ -180,6 +189,44 @@ def _iter_records(registry_dir: str):
|
||||
logger.warning("skipping unreadable record %s: %s", path, exc)
|
||||
|
||||
|
||||
def get_feedback(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> str:
|
||||
"""Read the job's audit log or events log and return the detail of the last completed/error event."""
|
||||
# 1) Try the unified audit log first (ndjson) since it's written synchronously by the subscriber
|
||||
try:
|
||||
import mqtt_common
|
||||
logs_dir = mqtt_common.LOGS_DIR
|
||||
events = list(mqtt_common.iter_logged_events(job_id, logs_dir))
|
||||
for e in reversed(events):
|
||||
if e.get("source_event") in ("completed", "error"):
|
||||
return e.get("detail", "")
|
||||
if e.get("event") in ("completed", "error"):
|
||||
return e.get("detail", "")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 2) Fallback to local .events.log
|
||||
log_path = Path(registry_dir) / f"{job_id}.events.log"
|
||||
if log_path.exists():
|
||||
feedback = ""
|
||||
try:
|
||||
with open(log_path, "r", encoding="utf-8") as fh:
|
||||
for line in fh:
|
||||
if not line.strip():
|
||||
continue
|
||||
try:
|
||||
payload = json.loads(line)
|
||||
if payload.get("event") in ("completed", "error"):
|
||||
feedback = payload.get("detail", "")
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
except OSError:
|
||||
pass
|
||||
if feedback:
|
||||
return feedback
|
||||
|
||||
return ""
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# CLI (so the bash wrapper can shell out without inline python)
|
||||
# --------------------------------------------------------------------------
|
||||
@@ -197,6 +244,10 @@ def _build_parser() -> argparse.ArgumentParser:
|
||||
p_reg.add_argument("--bits", type=int, default=32, help="32 (PoC) or 128 (prod)")
|
||||
p_reg.add_argument("--artifact", action="append", default=[], dest="artifacts")
|
||||
p_reg.add_argument("--auth-token", default=None, help="HMAC auth token for the job (auto-generated if secure broker is detected)")
|
||||
p_reg.add_argument("--job-type", default="direct", choices=["direct", "loop", "discuss"])
|
||||
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_list = sub.add_parser("list", help="list jobs (optionally by status)")
|
||||
p_list.add_argument("--status", default=None)
|
||||
@@ -209,6 +260,16 @@ def _build_parser() -> argparse.ArgumentParser:
|
||||
p_status.add_argument("--job", required=True)
|
||||
p_status.add_argument("--set", required=True, dest="status")
|
||||
|
||||
p_update = sub.add_parser("update", help="update a job record")
|
||||
p_update.add_argument("--job", required=True)
|
||||
p_update.add_argument("--status", default=None)
|
||||
p_update.add_argument("--agent-session", default=None)
|
||||
p_update.add_argument("--prompt", default=None)
|
||||
p_update.add_argument("--iteration", type=int, default=None)
|
||||
|
||||
p_feedback = sub.add_parser("get-feedback", help="get the last feedback detail (completed/error) for a job")
|
||||
p_feedback.add_argument("--job", required=True)
|
||||
|
||||
p_pick = sub.add_parser("pick", help="claim a pending job for a session; prints id")
|
||||
p_pick.add_argument("--agent-session", default="tmux:claude")
|
||||
|
||||
@@ -247,6 +308,10 @@ def main(argv: Optional[List[str]] = None) -> int:
|
||||
expected_artifacts=args.artifacts,
|
||||
bits=args.bits,
|
||||
auth_token=args.auth_token,
|
||||
job_type=args.job_type,
|
||||
reviewer=args.reviewer,
|
||||
reviewer_session=args.reviewer_session,
|
||||
max_iterations=args.max_iterations,
|
||||
)
|
||||
print(job_id)
|
||||
return 0
|
||||
@@ -279,6 +344,27 @@ def main(argv: Optional[List[str]] = None) -> int:
|
||||
return 1
|
||||
return 0
|
||||
|
||||
if args.command == "update":
|
||||
fields = {}
|
||||
if args.status is not None:
|
||||
fields["status"] = args.status
|
||||
if args.agent_session is not None:
|
||||
fields["agent_session"] = args.agent_session
|
||||
if args.prompt is not None:
|
||||
fields["prompt"] = args.prompt
|
||||
if args.iteration is not None:
|
||||
fields["iteration"] = args.iteration
|
||||
try:
|
||||
mqtt_common.update_job_status(args.job, rd, **fields)
|
||||
except FileNotFoundError as exc:
|
||||
print(str(exc), file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
if args.command == "get-feedback":
|
||||
print(get_feedback(args.job, rd))
|
||||
return 0
|
||||
|
||||
if args.command == "pick":
|
||||
job_id = pick_pending(args.agent_session, rd)
|
||||
if job_id is None:
|
||||
|
||||
Reference in New Issue
Block a user