Compare commits

...
4 Commits
Author SHA1 Message Date
GodopuandClaude Fable 5.1 a7cb90e63a fix(delegation): the sibling rule pairs with the registry's real path (Issue #57, Mac R2)
A registry directory that is itself a symlink got two audit logs: submit
resolved --registry-dir with pwd -P and wrote next to the target, while logs
derived the sibling from the link path. sibling_logs_dir() now takes the
registry's realpath; an explicit DELEGATE_JOB_LOGS_DIR still wins.
Also: forwarded_keys() normalises both sides of its comparison, and the
design doc's status row lists the audit log next to the registry.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-15 00:06:30 +00:00
GodopuandClaude Fable 5.1 1c598bb546 fix(subscriber): the disk fallback re-reads the record before taking an audit status (Issue #57, Mac R1)
The record (status, token, attempt, transition_seq) was read before the
audit status.json; if A error -> redispatch (attempt 2) -> B completed
happened in between, the audit terminal was B's while the token and attempt
validated were A's, and A's caller was told "completed". The audit status
is now taken only when a fresh read of the record still shows the same
token, attempt and transition_seq; otherwise the job stays pending and the
next round reads both afresh. The no-record case is unchanged.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-15 00:04:52 +00:00
GodopuandClaude Fable 5.1 151af99765 fix(delegate-job): logs passes --registry-dir on only when given (Issue #57, Mac F2)
cmd_logs always passed the wrapper's cwd-relative default registry, so the
usual run (registry not moved, cwd not the workspace) no longer found its
jobs. Without the option registry.py's own default decides, as before:
MAM_JOBS_DIR, then WORKSPACE_ROOT, then the cwd.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-15 00:02:50 +00:00
GodopuandClaude Fable 5.1 75c004222c fix(delegation): the path rules live in a stdlib-only module; the package imports without paho (Issue #57, Mac F1)
control_plane and status.sh imported sibling_logs_dir from mqtt_common,
which imports paho at module level, so create/resume's binding and status
died on an interpreter without paho-mqtt (_delegate_py_bin's last fallback
is the PATH's python3). The registry/audit-log path rules move to
lib_py/delegation/logroot.py (standard library only) and mqtt_common
re-exports them. The package's own import chain also reached paho through
service -> publish_event -> publisher, so mqtt_common and publisher import
paho tolerantly and make_client() refuses with a clear error when a broker
is actually needed. control_plane, status.sh, lib.sh's delegate_publish_cmd
and delegate_logs_dir, and the wrapper's _logs_dir_abs import from logroot.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-15 00:02:42 +00:00
10 changed files with 279 additions and 66 deletions

No files matched your search

+2 -2
View File
@@ -2322,7 +2322,7 @@ delegate_publish_cmd() {
pub_reg="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:+$WORKSPACE_ROOT/.mam/jobs}}" pub_reg="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:+$WORKSPACE_ROOT/.mam/jobs}}"
pub_reg="${pub_reg:-.mam/jobs}" pub_reg="${pub_reg:-.mam/jobs}"
mkdir -p "$pub_reg" && pub_reg="$(cd -P "$pub_reg" && pwd -P)" mkdir -p "$pub_reg" && pub_reg="$(cd -P "$pub_reg" && pwd -P)"
pub_logs="$(PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$py_bin" -c 'import os, sys; from lib_py.delegation.mqtt_common import get_logs_dir; print(os.path.abspath(get_logs_dir(sys.argv[1])))' "$pub_reg" 2>/dev/null || true)" pub_logs="$(PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$py_bin" -c 'import os, sys; from lib_py.delegation.logroot import get_logs_dir; print(os.path.abspath(get_logs_dir(sys.argv[1])))' "$pub_reg" 2>/dev/null || true)"
local -a argv=("$py_bin" "$pub_script" --registry-dir "$pub_reg" --job "$job_id") local -a argv=("$py_bin" "$pub_script" --registry-dir "$pub_reg" --job "$job_id")
[ -n "$tok" ] && argv+=(--expect-dispatch-token "$tok") [ -n "$tok" ] && argv+=(--expect-dispatch-token "$tok")
[ -n "$att" ] && argv+=(--attempt-id "$att") [ -n "$att" ] && argv+=(--attempt-id "$att")
@@ -2373,7 +2373,7 @@ except Exception:
# closes it or another dispatch claims it. Lines are parsed as JSON, never # closes it or another dispatch claims it. Lines are parsed as JSON, never
# pattern-matched. rc 0 = recorded, 1 = not (yet). # pattern-matched. rc 0 = recorded, 1 = not (yet).
delegate_logs_dir() { delegate_logs_dir() {
PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -c 'import os, sys; from lib_py.delegation.mqtt_common import get_logs_dir; print(os.path.abspath(get_logs_dir(sys.argv[1])))' "$1" 2>/dev/null PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -c 'import os, sys; from lib_py.delegation.logroot import get_logs_dir; print(os.path.abspath(get_logs_dir(sys.argv[1])))' "$1" 2>/dev/null
} }
delegate_started_recorded() { delegate_started_recorded() {
+4 -3
View File
@@ -25,7 +25,7 @@ import shlex
import sys import sys
from typing import Dict, Optional from typing import Dict, Optional
from lib_py.delegation import mqtt_common from lib_py.delegation.logroot import sibling_logs_dir
KEYS = ("workspace_root", "agent_sessions_yaml", "mam_jobs_dir", "delegate_job_logs_dir") KEYS = ("workspace_root", "agent_sessions_yaml", "mam_jobs_dir", "delegate_job_logs_dir")
ENV = {"workspace_root": "WORKSPACE_ROOT", "agent_sessions_yaml": "AGENT_SESSIONS_YAML", ENV = {"workspace_root": "WORKSPACE_ROOT", "agent_sessions_yaml": "AGENT_SESSIONS_YAML",
@@ -54,7 +54,7 @@ def from_env(env=None) -> Dict[str, str]:
# Issue #57: without an explicit DELEGATE_JOB_LOGS_DIR the audit log is # Issue #57: without an explicit DELEGATE_JOB_LOGS_DIR the audit log is
# the delegate_job_logs next to the registry in use, not the workspace's. # the delegate_job_logs next to the registry in use, not the workspace's.
if not (env.get("DELEGATE_JOB_LOGS_DIR") or "").strip(): if not (env.get("DELEGATE_JOB_LOGS_DIR") or "").strip():
b["delegate_job_logs_dir"] = _abs(mqtt_common.sibling_logs_dir(b["mam_jobs_dir"])) b["delegate_job_logs_dir"] = _abs(sibling_logs_dir(b["mam_jobs_dir"]))
return b return b
@@ -79,7 +79,8 @@ def forwarded_keys(desc: dict) -> list:
when it is the sibling of the registry and so derivable there (Issue #57). when it is the sibling of the registry and so derivable there (Issue #57).
An explicit, non-sibling root is forwarded as before.""" An explicit, non-sibling root is forwarded as before."""
keys = list(KEYS) keys = list(KEYS)
if desc.get("delegate_job_logs_dir") == _abs(mqtt_common.sibling_logs_dir(desc["mam_jobs_dir"])): logs = desc.get("delegate_job_logs_dir")
if logs and _abs(logs) == _abs(sibling_logs_dir(desc["mam_jobs_dir"])):
keys.remove("delegate_job_logs_dir") keys.remove("delegate_job_logs_dir")
return keys return keys
@@ -118,7 +118,24 @@ def _check_disk_fallback(
logger.warning("disk fallback ignored audit status %r for %s: transition_seq %r is older " logger.warning("disk fallback ignored audit status %r for %s: transition_seq %r is older "
"than the record's %r", logged.get("status"), jid, aud_seq, rec_seq) "than the record's %r", logged.get("status"), jid, aud_seq, rec_seq)
else: else:
disk_status = logged.get("status") # The record above was read before the audit status. If
# another dispatch began in between (A error -> redispatch
# -> B completed), the audit terminal is B's while the token
# and attempt checked below are A's. Take it only when the
# record still describes the same observation; otherwise
# leave the job pending for the next round, which reads
# both afresh (Issue #57, Mac review R1).
try:
now = load_job(jid, registry_dir)
except Exception:
now = {}
same = ((now.get("dispatch_token"), now.get("attempt_id", now.get("iteration")),
now.get("transition_seq")) == (rec_token, rec_attempt, rec_seq))
if same:
disk_status = logged.get("status")
else:
logger.warning("disk fallback deferred %s: the record changed while the audit status "
"was being read", jid)
if disk_status in ("completed", "error", "cancelled"): if disk_status in ("completed", "error", "cancelled"):
if expected_dispatch_tokens: if expected_dispatch_tokens:
@@ -0,0 +1,70 @@
"""Where the job registry and its audit log live (Issue #15, Issue #57).
Standard library only, on purpose: create/resume's control-plane binding,
status.sh and lib.sh resolve these paths through ``_delegate_py_bin``, whose
last fallback is the PATH's ``python3``, which may not have paho-mqtt.
``mqtt_common`` re-exports every name here, so existing callers are unchanged.
"""
from __future__ import annotations
import os
from typing import Optional
DEFAULT_REGISTRY_DIR = ".mam/jobs"
def _workspace_mam_dir(leaf: str) -> Optional[str]:
"""`$WORKSPACE_ROOT/.mam/<leaf>` when lib.sh has told us which workspace we
serve; None otherwise."""
root = os.environ.get("WORKSPACE_ROOT")
if root and root.strip() and os.path.isdir(root):
return os.path.join(root, ".mam", leaf)
return None
def default_registry_dir() -> str:
"""Registry root, resolved at call time (Issue #15).
``MAM_JOBS_DIR`` wins; otherwise the workspace lib.sh exported as
``WORKSPACE_ROOT``; otherwise the historical cwd-relative ``.mam/jobs``.
Anchoring on the workspace is what keeps a test sandbox (or any caller
whose cwd is not the workspace) from registering jobs into whichever
``.mam/jobs`` happens to sit under the current directory - the pytest
suite used to leave hundreds of onboarding records in the real registry.
"""
env = os.environ.get("MAM_JOBS_DIR")
if env and env.strip():
return env
return _workspace_mam_dir("jobs") or DEFAULT_REGISTRY_DIR
def sibling_logs_dir(registry_dir: str) -> str:
"""The audit-log root that pairs with ``registry_dir``: the
``delegate_job_logs`` next to it. Pure; reads no environment. Issue #57.
The registry is taken by its real path: a registry directory that is itself
a symlink pairs with the audit log next to its target, which is what the
wrapper (``pwd -P``) and the control-plane binding (``realpath``) already
see, so submit and logs agree (Mac review R2). A relative registry resolves
against the cwd, as registry.py opens it.
"""
return os.path.join(os.path.dirname(os.path.realpath(registry_dir)), "delegate_job_logs")
def get_logs_dir(registry_dir: Optional[str] = None) -> str:
"""Audit-log root, resolved at call time (B-9).
``DELEGATE_JOB_LOGS_DIR`` when set: it means the audit log was moved to a
place that cannot be derived from the registry. Otherwise the audit log
follows the registry (Issue #57): the ``delegate_job_logs`` next to
``registry_dir``, or next to ``default_registry_dir()`` when the caller
does not know its registry. The default registry is
``$WORKSPACE_ROOT/.mam/jobs`` (Issue #15), otherwise ``<cwd>/.mam/jobs``,
so the usual roots are unchanged. Resolved per call rather than at import
so a chdir after import cannot strand the audit trail in the old tree —
the same reason ``DEFAULT_REGISTRY_DIR`` stays a relative string.
"""
env = os.environ.get("DELEGATE_JOB_LOGS_DIR")
if env and env.strip():
return env
return sibling_logs_dir(registry_dir or default_registry_dir())
@@ -30,9 +30,14 @@ from dataclasses import asdict, dataclass
from pathlib import Path from pathlib import Path
from typing import Any, Callable, Dict, Iterable, List, Optional from typing import Any, Callable, Dict, Iterable, List, Optional
import paho.mqtt.client as mqtt try:
import paho.mqtt.client as mqtt
except ImportError: # the registry and the path rules work without a broker library (Issue #57)
mqtt = None # make_client() says so when a broker is actually needed
from lib_py.delegation import envfile from lib_py.delegation import envfile
from lib_py.delegation.logroot import ( # noqa: F401 (re-exported: this module is the documented home)
DEFAULT_REGISTRY_DIR, _workspace_mam_dir, default_registry_dir, get_logs_dir, sibling_logs_dir)
logger = logging.getLogger("delegate_job.mqtt_common") logger = logging.getLogger("delegate_job.mqtt_common")
@@ -145,35 +150,10 @@ def load_env_or_report() -> bool:
# Constants # Constants
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
SCHEMA_VERSION = 1 SCHEMA_VERSION = 1
DEFAULT_REGISTRY_DIR = ".mam/jobs"
DEFAULT_TOPIC_ROOT = "python/mqtt/jobs" DEFAULT_TOPIC_ROOT = "python/mqtt/jobs"
LOCK_FILENAME = ".lock" LOCK_FILENAME = ".lock"
def _workspace_mam_dir(leaf: str) -> Optional[str]:
"""`$WORKSPACE_ROOT/.mam/<leaf>` when lib.sh has told us which workspace we
serve; None otherwise."""
root = os.environ.get("WORKSPACE_ROOT")
if root and root.strip() and os.path.isdir(root):
return os.path.join(root, ".mam", leaf)
return None
def default_registry_dir() -> str:
"""Registry root, resolved at call time (Issue #15).
``MAM_JOBS_DIR`` wins; otherwise the workspace lib.sh exported as
``WORKSPACE_ROOT``; otherwise the historical cwd-relative ``.mam/jobs``.
Anchoring on the workspace is what keeps a test sandbox (or any caller
whose cwd is not the workspace) from registering jobs into whichever
``.mam/jobs`` happens to sit under the current directory - the pytest
suite used to leave hundreds of onboarding records in the real registry.
"""
env = os.environ.get("MAM_JOBS_DIR")
if env and env.strip():
return env
return _workspace_mam_dir("jobs") or DEFAULT_REGISTRY_DIR
# Persistent audit-log layout: .mam/delegate_job_logs/<job_id>/{meta,events,status}. # Persistent audit-log layout: .mam/delegate_job_logs/<job_id>/{meta,events,status}.
# This is a *separate* artifact from the registry: the registry is the live job # This is a *separate* artifact from the registry: the registry is the live job
# record (mutated in place), the audit log is an append-only history that # record (mutated in place), the audit log is an append-only history that
@@ -183,31 +163,6 @@ EVENTS_FILENAME = "events.ndjson"
STATUS_FILENAME = "status.json" STATUS_FILENAME = "status.json"
def sibling_logs_dir(registry_dir: str) -> str:
"""The audit-log root that pairs with ``registry_dir``: its sibling
``delegate_job_logs``. Pure; reads no environment. Issue #57."""
return os.path.join(os.path.dirname(os.path.abspath(registry_dir)), "delegate_job_logs")
def get_logs_dir(registry_dir: Optional[str] = None) -> str:
"""Audit-log root, resolved at call time (B-9).
``DELEGATE_JOB_LOGS_DIR`` when set: it means the audit log was moved to a
place that cannot be derived from the registry. Otherwise the audit log
follows the registry (Issue #57): the ``delegate_job_logs`` next to
``registry_dir``, or next to ``default_registry_dir()`` when the caller
does not know its registry. The default registry is
``$WORKSPACE_ROOT/.mam/jobs`` (Issue #15), otherwise ``<cwd>/.mam/jobs``,
so the usual roots are unchanged. Resolved per call rather than at import
so a chdir after import cannot strand the audit trail in the old tree —
the same reason ``DEFAULT_REGISTRY_DIR`` stays a relative string.
"""
env = os.environ.get("DELEGATE_JOB_LOGS_DIR")
if env and env.strip():
return env
return sibling_logs_dir(registry_dir or default_registry_dir())
def __getattr__(name: str): # PEP 562 (3.7+) def __getattr__(name: str): # PEP 562 (3.7+)
"""Keep ``mqtt_common.LOGS_DIR`` working for external consumers """Keep ``mqtt_common.LOGS_DIR`` working for external consumers
(documented in registry.md) while resolving it dynamically.""" (documented in registry.md) while resolving it dynamically."""
@@ -328,6 +283,9 @@ def make_client(role: str, config: Optional[BrokerConfig] = None) -> mqtt.Client
""" """
config = config or broker_config_from_env() config = config or broker_config_from_env()
client_id = f"{config.client_id_prefix}-{role}-{uuid.uuid4().hex[:8]}" client_id = f"{config.client_id_prefix}-{role}-{uuid.uuid4().hex[:8]}"
if mqtt is None:
raise ImportError("paho-mqtt is required to talk to a broker; install it into the virtualenv "
"(see multi-agent-mux-delegate-job/requirements.txt)")
client = mqtt.Client( client = mqtt.Client(
callback_api_version=mqtt.CallbackAPIVersion.VERSION2, callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
client_id=client_id, client_id=client_id,
@@ -14,7 +14,10 @@ import logging
import time import time
from typing import Any, Optional, Tuple from typing import Any, Optional, Tuple
import paho.mqtt.client as mqtt try:
import paho.mqtt.client as mqtt
except ImportError: # importable without a broker library; mqtt_common.make_client() refuses to connect
mqtt = None
from lib_py.delegation import mqtt_common from lib_py.delegation import mqtt_common
@@ -92,7 +92,7 @@ _abs_registry_dir() {
# #57), pinned into the injected commands so an agent publishing from another # #57), pinned into the injected commands so an agent publishing from another
# cwd does not fall back to its own default. # cwd does not fall back to its own default.
_logs_dir_abs() { _logs_dir_abs() {
"$PY" -c 'import os, sys; sys.path.insert(0, sys.argv[1]); from lib_py.delegation.mqtt_common import get_logs_dir; print(os.path.abspath(get_logs_dir(sys.argv[2])))' "$SCRIPT_DIR/.." "$REGISTRY_DIR" "$PY" -c 'import os, sys; sys.path.insert(0, sys.argv[1]); from lib_py.delegation.logroot import get_logs_dir; print(os.path.abspath(get_logs_dir(sys.argv[2])))' "$SCRIPT_DIR/.." "$REGISTRY_DIR"
} }
# Issue #17 item 4: the wrapper's decisions are lib_py/delegation/wrapper.py; # Issue #17 item 4: the wrapper's decisions are lib_py/delegation/wrapper.py;
@@ -1093,19 +1093,24 @@ cmd_logs() {
# (Issue #57; $DELEGATE_JOB_LOGS_DIR overrides). parse_opts is for submit's # (Issue #57; $DELEGATE_JOB_LOGS_DIR overrides). parse_opts is for submit's
# globals, so the few options are read here. # globals, so the few options are read here.
PY="$(pick_python)" PY="$(pick_python)"
local reg="$REGISTRY_DIR_DEFAULT" jid="" list=0 # --registry-dir is passed on only when given: registry.py's own default
# follows MAM_JOBS_DIR, then WORKSPACE_ROOT, so the usual run works from any
# cwd (the wrapper's REGISTRY_DIR_DEFAULT is cwd-relative). bash 3.2 + set -u:
# an empty array expands through ${reg[@]+"${reg[@]}"}.
local -a reg=()
local jid="" list=0
while [[ $# -gt 0 ]]; do while [[ $# -gt 0 ]]; do
case "$1" in case "$1" in
--registry-dir) reg="$2"; shift 2;; --registry-dir) reg=(--registry-dir "$2"); shift 2;;
--list) list=1; shift;; --list) list=1; shift;;
*) jid="$1"; shift;; *) jid="$1"; shift;;
esac esac
done done
if [[ "$list" == "1" ]]; then if [[ "$list" == "1" ]]; then
"$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$reg" logs --list "$PY" "$SCRIPT_DIR/scripts/registry.py" ${reg[@]+"${reg[@]}"} logs --list
else else
[[ -n "$jid" ]] || { echo "logs requires <job_id> or --list" >&2; exit 1; } [[ -n "$jid" ]] || { echo "logs requires <job_id> or --list" >&2; exit 1; }
"$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$reg" logs "$jid" "$PY" "$SCRIPT_DIR/scripts/registry.py" ${reg[@]+"${reg[@]}"} logs "$jid"
fi fi
} }
@@ -168,7 +168,7 @@ def get_job_status(s):
candidates += [os.path.join(r, jid, 'status.json') candidates += [os.path.join(r, jid, 'status.json')
for r in (os.environ.get('DELEGATE_JOB_LOGS_DIR'), cp.get('delegate_job_logs_dir')) if r] for r in (os.environ.get('DELEGATE_JOB_LOGS_DIR'), cp.get('delegate_job_logs_dir')) if r]
# Issue #57: the audit log lives next to the registry unless moved explicitly # Issue #57: the audit log lives next to the registry unless moved explicitly
from lib_py.delegation.mqtt_common import sibling_logs_dir from lib_py.delegation.logroot import sibling_logs_dir
candidates += [os.path.join(sibling_logs_dir(r), jid, 'status.json') candidates += [os.path.join(sibling_logs_dir(r), jid, 'status.json')
for r in (os.environ.get('MAM_JOBS_DIR'), cp.get('mam_jobs_dir')) if r] for r in (os.environ.get('MAM_JOBS_DIR'), cp.get('mam_jobs_dir')) if r]
candidates += [ candidates += [
@@ -380,7 +380,7 @@ def get_job_status(s):
candidates += [os.path.join(r, jid, 'status.json') candidates += [os.path.join(r, jid, 'status.json')
for r in (os.environ.get('DELEGATE_JOB_LOGS_DIR'), cp.get('delegate_job_logs_dir')) if r] for r in (os.environ.get('DELEGATE_JOB_LOGS_DIR'), cp.get('delegate_job_logs_dir')) if r]
# Issue #57: the audit log lives next to the registry unless moved explicitly # Issue #57: the audit log lives next to the registry unless moved explicitly
from lib_py.delegation.mqtt_common import sibling_logs_dir from lib_py.delegation.logroot import sibling_logs_dir
candidates += [os.path.join(sibling_logs_dir(r), jid, 'status.json') candidates += [os.path.join(sibling_logs_dir(r), jid, 'status.json')
for r in (os.environ.get('MAM_JOBS_DIR'), cp.get('mam_jobs_dir')) if r] for r in (os.environ.get('MAM_JOBS_DIR'), cp.get('mam_jobs_dir')) if r]
candidates += [ candidates += [
+1 -1
View File
@@ -126,7 +126,7 @@ bash "$MAIN/.agents/skills/multi-agent-mux-stop/scripts/worktree_remove.sh" --pa
| create | env 에서 결정 → 행 `control_plane` 에 항상 영속. `--workspace` 기본값과 다르면 새 pane 의 생성 호출(`pane split`/`workspace create`)에 바인딩(유도 가능한 감사 로그 루트 제외)을 `--env` 로 전파한다. 전파가 필요하면 idle pane 을 재사용하지 않는다(`agent start -- <args>` 는 `--` 뒤를 에이전트 인자로 넘기므로 `env K=V` 를 쓸 수 없다). 전파가 필요한데 경로에 공백이 있으면 exit 2. `~/.local/bin` wrapper 경로는 기록만 하고 경고 | pane.cwd = `--workspace`. 바인딩 계산 실패 시 워크트리면 exit 2, 제어 루트면 경고 | — (루프 A: 즉시 reconcile·watchdog 은 제어 루트 코드와 `WORKSPACE_ROOT` 로 기동, prepare 가 `MAM_NO_SUBSCRIBER_DAEMON=1`) | | create | env 에서 결정 → 행 `control_plane` 에 항상 영속. `--workspace` 기본값과 다르면 새 pane 의 생성 호출(`pane split`/`workspace create`)에 바인딩(유도 가능한 감사 로그 루트 제외)을 `--env` 로 전파한다. 전파가 필요하면 idle pane 을 재사용하지 않는다(`agent start -- <args>` 는 `--` 뒤를 에이전트 인자로 넘기므로 `env K=V` 를 쓸 수 없다). 전파가 필요한데 경로에 공백이 있으면 exit 2. `~/.local/bin` wrapper 경로는 기록만 하고 경고 | pane.cwd = `--workspace`. 바인딩 계산 실패 시 워크트리면 exit 2, 제어 루트면 경고 | — (루프 A: 즉시 reconcile·watchdog 은 제어 루트 코드와 `WORKSPACE_ROOT` 로 기동, prepare 가 `MAM_NO_SUBSCRIBER_DAEMON=1`) |
| resume | 행 바인딩 우선(호출자 env 와 다르면 stderr 경고) 재전파. 바인딩 없는 기존 행은 호출자 바인딩으로 채움. 이미 살아 있는 세션에도 바인딩을 영속한다. 계산 실패 시 워크트리면 exit 2 | 원래 cwd 로만: 경로 불일치·부재 시 exit 2 | 워크트리 삭제 시 재개 불가(대화 보존, 같은 경로에 다시 만들면 재개) | | resume | 행 바인딩 우선(호출자 env 와 다르면 stderr 경고) 재전파. 바인딩 없는 기존 행은 호출자 바인딩으로 채움. 이미 살아 있는 세션에도 바인딩을 영속한다. 계산 실패 시 워크트리면 exit 2 | 원래 cwd 로만: 경로 불일치·부재 시 exit 2 | 워크트리 삭제 시 재개 불가(대화 보존, 같은 경로에 다시 만들면 재개) |
| stop | `MAM_JOBS_DIR` > 행의 `mam_jobs_dir` > `$WORKSPACE_ROOT/.mam/jobs` 순으로 잡 조회·발행. 감사 로그도 `DELEGATE_JOB_LOGS_DIR` > 행의 `delegate_job_logs_dir` 순 | 행 cwd 로 uuid 캡처·purge. 워크트리 해체는 `worktree_remove.sh` 가 판정(잠금·세션·잡·청결·서브모듈) 뒤에만 정지·제거, 서브모듈 워크트리는 검증 후 `--force` 1회 | — | | stop | `MAM_JOBS_DIR` > 행의 `mam_jobs_dir` > `$WORKSPACE_ROOT/.mam/jobs` 순으로 잡 조회·발행. 감사 로그도 `DELEGATE_JOB_LOGS_DIR` > 행의 `delegate_job_logs_dir` 순 | 행 cwd 로 uuid 캡처·purge. 워크트리 해체는 `worktree_remove.sh` 가 판정(잠금·세션·잡·청결·서브모듈) 뒤에만 정지·제거, 서브모듈 워크트리는 검증 후 `--force` 1회 | — |
| status | `MAM_JOBS_DIR`·행 `mam_jobs_dir`·감사 로그(`DELEGATE_JOB_LOGS_DIR`·행 `delegate_job_logs_dir`)·기존 후보 순 조회 | 행 cwd 로 대화 키 | — | | status | `MAM_JOBS_DIR`·행 `mam_jobs_dir`·감사 로그(`DELEGATE_JOB_LOGS_DIR`·행 `delegate_job_logs_dir`)·레지스트리 옆 감사 로그(`MAM_JOBS_DIR`·행 `mam_jobs_dir` 의 형제 `delegate_job_logs`, #57)·기존 후보 순 조회 | 행 cwd 로 대화 키 | — |
| monitor/reconcile | `MAM_JOBS_DIR` 우선(정적 가드로만 검증) | 기존 행은 행 cwd 로 검증 | 루트 밖 cwd 의 미등록 세션은 자동 등록 안 됨 → create/prepare 로만 등록 (남은 제한) | | monitor/reconcile | `MAM_JOBS_DIR` 우선(정적 가드로만 검증) | 기존 행은 행 cwd 로 검증 | 루트 밖 cwd 의 미등록 세션은 자동 등록 안 됨 → create/prepare 로만 등록 (남은 제한) |
| delegate-job | `--registry-dir` > `MAM_JOBS_DIR` > `.mam/jobs`(`--workdir` 기준), 모두 절대화. 발행 명령이 감사 로그 루트(`DELEGATE_JOB_LOGS_DIR`)를 고정 | 문구·명령·리포트 경로 절대 | 작은따옴표가 든 경로는 지원 범위 밖 | | delegate-job | `--registry-dir` > `MAM_JOBS_DIR` > `.mam/jobs`(`--workdir` 기준), 모두 절대화. 발행 명령이 감사 로그 루트(`DELEGATE_JOB_LOGS_DIR`)를 고정 | 문구·명령·리포트 경로 절대 | 작은따옴표가 든 경로는 지원 범위 밖 |
| loop | `MAM_JOBS_DIR` 존중, 래퍼에 `--registry-dir` 명시. `--worktree` 로 소스 루트 분리 | diff·base·head 만 소스 루트. **구현이 루프 밖이면 `--base <구현 전 sha> --review-only` 필수**. `--worktree`/`--base` 가 있으면 빈 변경분 거부 | 루프 하나(8). 원격 리뷰어는 identity 모드에서 거부(기본) 또는 diff-only(`--remote-review diff-only`); 원격 리뷰어가 트리를 검증하게 하는 스냅샷 공유(공유 원격의 `refs/mam/review/<loop_id>/<n>`)는 범위 밖. 레거시 셸 루프(`MAM_LOOP_PY=0`)는 새 옵션 거부 | | loop | `MAM_JOBS_DIR` 존중, 래퍼에 `--registry-dir` 명시. `--worktree` 로 소스 루트 분리 | diff·base·head 만 소스 루트. **구현이 루프 밖이면 `--base <구현 전 sha> --review-only` 필수**. `--worktree`/`--base` 가 있으면 빈 변경분 거부 | 루프 하나(8). 원격 리뷰어는 identity 모드에서 거부(기본) 또는 diff-only(`--remote-review diff-only`); 원격 리뷰어가 트리를 검증하게 하는 스냅샷 공유(공유 원격의 `refs/mam/review/<loop_id>/<n>`)는 범위 밖. 레거시 셸 루프(`MAM_LOOP_PY=0`)는 새 옵션 거부 |
+159
View File
@@ -472,3 +472,162 @@ def test_a_forwarded_pane_moves_its_logs_with_the_registry(tmp_path, monkeypatch
assert res.stdout.strip().splitlines()[-1] == str(Path(os.path.realpath(main)) / ".mam" / "delegate_job_logs" / jid2) assert res.stdout.strip().splitlines()[-1] == str(Path(os.path.realpath(main)) / ".mam" / "delegate_job_logs" / jid2)
assert os.listdir(real_logs) == [jid2] assert os.listdir(real_logs) == [jid2]
assert not (wt / ".mam" / "delegate_job_logs").exists() assert not (wt / ".mam" / "delegate_job_logs").exists()
# ---------------------------------------------------------------------------
# Mac review F1: the path helpers must not need paho
# ---------------------------------------------------------------------------
@pytest.fixture
def nopaho(tmp_path):
"""A `paho` package that fails to import, first on PYTHONPATH: the real
import path then behaves like an interpreter without paho-mqtt."""
d = tmp_path / "nopaho"
(d / "paho").mkdir(parents=True)
(d / "paho" / "__init__.py").write_text("raise ModuleNotFoundError(\"No module named 'paho'\")\n")
probe = subprocess.run([sys.executable, "-c", "import paho.mqtt.client"], capture_output=True, text=True,
env={**os.environ, "PYTHONPATH": f"{d}:{SKILLS}"}, timeout=60)
assert probe.returncode != 0 and "No module named 'paho'" in probe.stderr, probe.stderr
return d
def test_control_plane_binding_works_without_paho(moved, nopaho):
"""_delegate_py_bin's last fallback is the PATH's python3, which may lack
paho: create/resume's binding and lib.sh's publish-command prefix must still
resolve the audit-log root."""
ws, reg, logs = moved
env = _env(ws, MAM_JOBS_DIR=str(reg), PYTHONPATH=f"{nopaho}:{SKILLS}", AGENT_PYTHON_BIN=sys.executable)
res = subprocess.run([sys.executable, "-m", "lib_py.control_plane", "json", str(ws / "wt")],
capture_output=True, text=True, env=env, cwd=str(ws), timeout=60)
assert res.returncode == 0, res.stderr
desc = json.loads(res.stdout)
assert desc["delegate_job_logs_dir"] == os.path.realpath(logs)
res = subprocess.run([sys.executable, "-m", "lib_py.control_plane", "prefix", json.dumps(desc)],
capture_output=True, text=True, env=env, cwd=str(ws), timeout=60)
assert res.returncode == 0 and "MAM_JOBS_DIR=" in res.stdout and "DELEGATE_JOB_LOGS_DIR=" not in res.stdout, res.stdout
# lib.sh's injected publish command still pins the root (it resolved it with the same interpreter)
res = subprocess.run(["bash", "-c", f"source '{SKILLS}/lib.sh' && delegate_publish_cmd j57"],
capture_output=True, text=True, env=env, cwd=str(ws), timeout=60)
assert res.returncode == 0, res.stderr
assert res.stdout.startswith(f"DELEGATE_JOB_LOGS_DIR={os.path.realpath(logs)} "), res.stdout + res.stderr
@pytest.mark.parametrize("mode", ["json", "text"])
def test_status_works_without_paho(mam_sandbox, nopaho, monkeypatch, mode):
tmp = mam_sandbox
reg = tmp / ".mam" / "jobs"
jid = registry.register_job("s57 nopaho", registry_dir=str(reg), agent_session="herdr:s57-nopaho")
registry.update_status(jid, str(reg), "completed")
_mutate(tmp, f"""
d['herdr_sessions'] = [{{'name': 's57-creator-claude', 'status': 'running', 'delegate_job_id': {jid!r},
'pane': {{'cwd': {str(tmp)!r}}}}}]
""")
monkeypatch.setenv("PYTHONPATH", str(nopaho))
monkeypatch.setenv("AGENT_PYTHON_BIN", sys.executable)
args = ["--json"] if mode == "json" else []
res = subprocess.run(["bash", _skill(tmp, "multi-agent-mux-status", "scripts", "status.sh"), *args],
capture_output=True, text=True, timeout=120)
assert res.returncode == 0, res.stdout[-500:] + res.stderr[-800:]
if mode == "json":
assert json.loads(res.stdout)["sessions_detail"][0]["job_status"] == "completed"
else:
(line,) = [l for l in res.stdout.splitlines() if l.startswith("s57-creator-claude")]
assert "completed" in line, line
# ---------------------------------------------------------------------------
# Mac review F2: `logs` without --registry-dir follows the workspace, not the cwd
# ---------------------------------------------------------------------------
def test_logs_without_registry_dir_follows_the_workspace_from_any_cwd(tmp_path, monkeypatch):
for k in DROP:
monkeypatch.delenv(k, raising=False)
ws, elsewhere = tmp_path / "ws", tmp_path / "elsewhere"
(ws / ".mam").mkdir(parents=True)
elsewhere.mkdir()
monkeypatch.setenv("WORKSPACE_ROOT", str(ws))
res = subprocess.run([sys.executable, str(REGISTRY_PY), "register", "--prompt", "f2", "--agent", "human",
"--agent-session", "herdr:f2"], capture_output=True, text=True, env=_env(ws),
cwd=str(elsewhere), timeout=60)
assert res.returncode == 0, res.stderr
jid = res.stdout.strip()
assert (ws / ".mam" / "jobs" / f"{jid}.json").is_file() and (ws / ".mam" / "delegate_job_logs" / jid).is_dir()
res = _wrapper(ws, "logs", "--list", cwd=elsewhere)
assert res.returncode == 0 and jid in res.stdout, res.stdout + res.stderr
res = _wrapper(ws, "logs", jid, cwd=elsewhere)
assert res.returncode == 0 and "registered" in res.stdout, res.stdout + res.stderr
assert not (elsewhere / ".mam").exists()
# ---------------------------------------------------------------------------
# Mac review R1: an audit terminal of a later dispatch is not the earlier one's success
# ---------------------------------------------------------------------------
def test_disk_fallback_rejects_a_terminal_that_belongs_to_a_later_dispatch(moved, monkeypatch):
"""Between the registry read and the audit read: A error -> redispatch
(attempt 2) -> B completed. The caller waiting on A must not be told
"completed"; the caller of B must."""
ws, reg, logs = moved
jid = registry.register_job("r1", registry_dir=str(reg), agent_session="herdr:t57")
assert registry.pick_pending("herdr:t57", registry_dir=str(reg)) == jid # claim: running, token A
rec = mqtt_common.load_job(jid, str(reg))
tok_a = rec["dispatch_token"]
assert rec["status"] == "running" and rec["attempt_id"] == 1
read, fired = mqtt_common.read_logged_status, []
def interleave(job, logs_dir=None):
if not fired: # once, right between the two reads, with real transitions
fired.append(1)
assert engine.transition(jid, "error", registry_dir=str(reg), source="operator").committed
assert engine.transition(jid, "redispatch", registry_dir=str(reg), source="operator",
fields={"attempt_id": 2}).committed
assert engine.transition(jid, "completed", registry_dir=str(reg), source="operator").committed
return read(job, logs_dir)
monkeypatch.setattr(mqtt_common, "read_logged_status", interleave)
pending, terminal = {jid}, {}
job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: tok_a}, {jid: 1})
assert fired, "the interleaving did not run"
assert pending == {jid} and terminal == {}, "B's completion was attributed to A"
rec = mqtt_common.load_job(jid, str(reg))
tok_b = rec["dispatch_token"]
assert tok_b != tok_a and rec["attempt_id"] == 2 and rec["status"] == "completed"
# the next round, and the caller of B, see the truth
pending, terminal = {jid}, {}
job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: tok_b}, {jid: 2})
assert pending == set() and terminal == {jid: "completed"}
pending, terminal = {jid}, {}
job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: tok_a}, {jid: 1})
assert pending == {jid} and terminal == {}
# ---------------------------------------------------------------------------
# Mac review R2: a registry directory that is itself a symlink has one audit log
# ---------------------------------------------------------------------------
def test_a_symlinked_registry_dir_resolves_to_one_audit_log(tmp_path, monkeypatch):
"""submit resolves --registry-dir with pwd -P, so its audit log lands next
to the link's target; the resolver must pair the same place for logs."""
from lib_py.delegation import logroot
for k in DROP:
monkeypatch.delenv(k, raising=False)
ws = tmp_path / "ws"
(ws / ".mam").mkdir(parents=True)
monkeypatch.setenv("WORKSPACE_ROOT", str(ws))
actual = tmp_path / "actual" / "jobs"
actual.mkdir(parents=True)
alias = tmp_path / "alias"
alias.mkdir()
(alias / "jobs").symlink_to(actual)
real_actual = Path(os.path.realpath(tmp_path / "actual"))
assert logroot.sibling_logs_dir(str(alias / "jobs")) == str(real_actual / "delegate_job_logs")
res = _wrapper(ws, "submit", "--dry-run", "--agent", "human", "--registry-dir", str(alias / "jobs"),
"--workdir", str(ws), "--prompt", "t")
assert res.returncode == 0, res.stdout[-800:] + res.stderr[-800:]
jid = _only_job_id(actual)
assert res.stdout.strip().splitlines()[-1] == str(real_actual / "delegate_job_logs" / jid)
assert (real_actual / "delegate_job_logs" / jid / "meta.json").is_file()
assert not (tmp_path / "alias" / "delegate_job_logs").exists()
res = _wrapper(ws, "logs", "--registry-dir", str(alias / "jobs"), jid)
assert res.returncode == 0 and "registered" in res.stdout, res.stdout + res.stderr
assert not (ws / ".mam" / "delegate_job_logs").exists()