209 lines
8.4 KiB
Markdown
209 lines
8.4 KiB
Markdown
# Job Registry
|
|
|
|
The registry is the **single source of truth** for delegated work. Job metadata
|
|
(id, prompt, broker, status, timeouts) lives in files, **not** environment
|
|
variables — so one herdr session can handle many jobs sequentially or in
|
|
parallel without collisions, and `publish_event.py` / `job_subscriber.py` can
|
|
reconstruct everything they need from the registry alone.
|
|
|
|
Reference implementation: [`./scripts/registry.py`](./scripts/registry.py)
|
|
(library + CLI) over the primitives in
|
|
[`./scripts/mqtt_common.py`](./scripts/mqtt_common.py).
|
|
|
|
---
|
|
|
|
## 1. Directory layout
|
|
|
|
```
|
|
.mam/jobs/
|
|
<job_id>.json # job metadata record (schema below)
|
|
<job_id>.events.log # append-only JSON-lines event log (debug, optional)
|
|
.lock # shared advisory lock (fcntl) for the whole registry
|
|
```
|
|
|
|
`registry_dir` defaults to `.mam/jobs` and is overridable everywhere via
|
|
`--registry-dir`.
|
|
|
|
---
|
|
|
|
## 2. Job record schema
|
|
|
|
```json
|
|
{
|
|
"schema_version": 1,
|
|
"job_id": "abc12345",
|
|
"status": "pending | running | completed | error | cancelled",
|
|
"created_at": "2026-06-19T09:30:00Z",
|
|
"updated_at": "2026-06-19T09:32:00Z",
|
|
"prompt": "정렬 문제 10개를 만들어 sort_problems.md로 저장…",
|
|
"agent": "claude-code",
|
|
"agent_session": "herdr:claude",
|
|
"broker": {
|
|
"host": "broker.hivemq.com",
|
|
"port": 1883,
|
|
"tls": false,
|
|
"username": null
|
|
},
|
|
"topic_prefix": "python/mqtt/jobs/abc12345",
|
|
"timeout_sec": 3600,
|
|
"idle_timeout_sec": 120,
|
|
"expected_artifacts": ["sort_problems.md"],
|
|
"last_seq": 0,
|
|
"auth_token": null
|
|
}
|
|
```
|
|
|
|
- `broker` lets `publish_event.py` connect from the record alone (env still
|
|
overrides toggles like `MQTT_TLS`). It never carries the password (Issue #12):
|
|
every sub-agent can read the record, so `MQTT_PASSWORD` is layered back from
|
|
the environment (`.mam.env`) at connect time. Records written before that rule
|
|
are cleaned on their next write; `registry.py scrub-secrets [--dry-run]` strips
|
|
every stored password from a whole registry at once (an empty `null` key left
|
|
by the old rule is not a secret; it goes on the record's next write). `auth_token` and `dispatch_token` stay in the record
|
|
because the agent's own `publish_event` needs them to sign and to pin its
|
|
attempt.
|
|
- `topic_prefix` → the events topic is `<topic_prefix>/events`.
|
|
- `last_seq` backs the monotonic `seq` counter so it survives process restarts.
|
|
- `expected_artifacts` is the hook a user `validate.sh` checks (existence/content).
|
|
- `auth_token` is `null` in PoC; production sets `secrets.token_urlsafe(32)`.
|
|
|
|
---
|
|
|
|
## 3. Concurrency rules
|
|
|
|
### PoC — fcntl advisory lock
|
|
|
|
Every read-modify-write (`register_job`, `pick_pending`, `update_status`,
|
|
`next_seq`) runs inside `registry_lock(registry_dir)`, an exclusive
|
|
`fcntl.flock` over `.lock`. Single-host, good enough for many herdr sessions on
|
|
one machine.
|
|
|
|
### Production — SQLite WAL
|
|
|
|
When delegation spans **multiple hosts**, the file lock no longer serialises
|
|
across machines. Migrate the same operations to a SQLite database in WAL mode
|
|
(`PRAGMA journal_mode=WAL`) with a transaction per claim. The function
|
|
signatures stay identical; only the storage backend changes.
|
|
|
|
---
|
|
|
|
## 4. How multiple sessions take only their own work
|
|
|
|
Each herdr session carries an `agent_session` label (`herdr:claude`,
|
|
`herdr:claude-a`, `herdr:claude-b`, …). `pick_pending(agent_session)`:
|
|
|
|
1. acquires the registry lock,
|
|
2. scans for the **oldest** record with `status == "pending"` **and**
|
|
matching `agent_session`,
|
|
3. flips it to `running` and writes it back **atomically**,
|
|
4. releases the lock and returns the `job_id` (or `None`).
|
|
|
|
Because the scan + flip happen under one lock, two sessions can never claim the
|
|
same job. Sessions with distinct labels naturally partition the work; sessions
|
|
sharing a label compete safely — first to acquire the lock wins, the other sees
|
|
the job already `running` and moves on.
|
|
|
|
```bash
|
|
# session A only ever runs its own pending jobs
|
|
PY scripts/registry.py pick --agent-session herdr:claude-a # prints id or exits 3
|
|
```
|
|
|
|
---
|
|
|
|
## 5. Atomic status updates
|
|
|
|
All writes use a temp-file + `os.replace` rename, which is atomic on POSIX:
|
|
|
|
1. take the registry lock,
|
|
2. load the current record,
|
|
3. mutate fields + refresh `updated_at` (and `last_seq` for `next_seq`),
|
|
4. write to `.<job_id>.<rand>.tmp` in the **same directory**, `fsync`,
|
|
5. `os.replace(tmp, <job_id>.json)`,
|
|
6. release the lock.
|
|
|
|
A reader therefore always sees either the old or the new complete record, never
|
|
a half-written file. This is the file-based equivalent of the rename trick
|
|
(`pending.<session>` → `running.<session>`) and maps cleanly onto a single
|
|
SQLite transaction when you migrate.
|
|
|
|
---
|
|
|
|
## 6. CLI quick reference
|
|
|
|
```bash
|
|
PY=.venv/bin/python
|
|
$PY scripts/registry.py register --prompt "…" --agent claude-code \
|
|
--agent-session herdr:claude --timeout 3600 --idle-timeout 120 # → prints job_id
|
|
$PY scripts/registry.py list # human table
|
|
$PY scripts/registry.py list --json # full records
|
|
$PY scripts/registry.py get --job <id> # one record
|
|
$PY scripts/registry.py status --job <id> --set completed # set status
|
|
$PY scripts/registry.py pick --agent-session herdr:claude # claim → running
|
|
```
|
|
|
|
Exit codes: `0` ok, `1` not found / bad status, `3` (`pick`) no pending job for
|
|
that session.
|
|
|
|
---
|
|
|
|
## 7. Persistent audit log
|
|
|
|
Separate from the registry, every job is also mirrored to a durable append-only
|
|
audit log at `.mam/delegate_job_logs/<job_id>/` (override with
|
|
`DELEGATE_JOB_LOGS_DIR`, an audit-log root that is **not** next to the
|
|
registry; default: the `delegate_job_logs` next to the registry in use). The registry
|
|
is **live state** mutated in place; the audit log is **history** that survives
|
|
even after the registry dir is cleaned up. It is git-ignored.
|
|
|
|
```
|
|
.mam/delegate_job_logs/<job_id>/
|
|
meta.json # registration snapshot (the full job record at register time)
|
|
events.ndjson # append-only, one JSON event per line, time-ordered
|
|
status.json # current status only (fast point-query)
|
|
```
|
|
|
|
`events.ndjson` lines are written automatically at four points:
|
|
|
|
| Trigger | line `event` | Source |
|
|
|---------|-------------|--------|
|
|
| `register_job` | `registered` | `registry.register_job` → `mqtt_common.init_job_log` |
|
|
| status change (`update_status`, `pick`, publish status sync) | `status_changed` (`from`/`to`) | `mqtt_common.update_job_status` / `pick_pending` |
|
|
| event published | `published` (embeds the exact payload) | `publish_event.py` |
|
|
| event received | `received` | `job_subscriber.py` |
|
|
|
|
Helpers live in [`./scripts/mqtt_common.py`](./scripts/mqtt_common.py):
|
|
`get_logs_dir` (audit-log root, resolved per call — a chdir after import no
|
|
longer strands the trail; `LOGS_DIR` remains as a dynamic compat alias),
|
|
`job_log_path`, `init_job_log`, `append_event` (fcntl-locked,
|
|
concurrent-append safe), `update_logged_status`, and the readers
|
|
`read_logged_meta` / `read_logged_status` / `iter_logged_events` /
|
|
`list_logged_jobs`. Every writer is **best-effort and isolated** — wrapped in
|
|
`try/except` with a `logger.warning`, so an audit-log failure never breaks the
|
|
registry write, the publish, or the subscribe it shadows.
|
|
|
|
Read them via the CLI:
|
|
|
|
```bash
|
|
PY=.venv/bin/python
|
|
$PY scripts/registry.py logs <job_id> # pretty timeline
|
|
$PY scripts/registry.py logs <job_id> --tail 20 # last 20 events
|
|
$PY scripts/registry.py logs <job_id> --json # raw JSON lines
|
|
$PY scripts/registry.py logs --list # every job, live status
|
|
```
|
|
|
|
---
|
|
|
|
> **구현 위치 각주 (v4.3.0)**: 본 문서가 인용하는 `scripts/*.py` 경로는 유효하며 계약도 불변이다. 실제 구현은 issue #5 phase 1에서 `.agents/skills/lib_py/delegation/`으로 이동했고, `scripts/*.py`는 그 패키지를 자기 옆 트리에 바인딩하는 호환 shim이다.
|
|
|
|
## Liveness contract fields (Issue #18)
|
|
|
|
| Field | Meaning | Default |
|
|
|---|---|---|
|
|
| `heartbeat_sec` | N: the agent publishes `progress` at least every N s (the `--heartbeat N` timer does it) | 30 |
|
|
| `heartbeat_misses` | K: missed heartbeats before the job is unresponsive | 4 |
|
|
| `idle_timeout_sec` | K x N unless `--idle-timeout` was given | 120 |
|
|
| `ack_timeout_sec` | T: `started` must arrive within T s of the injection | 180 |
|
|
| `transport` | `herdr` (pane injection) or `remote` (signed retained dispatch over the broker) | `herdr` |
|
|
|
|
`<registry>/<job>.heartbeat.pid` / `.heartbeat.log` belong to the running heartbeat timer; a terminal publish removes the pidfile.
|