Compare commits
4
Commits
63d816468e
...
a7cb90e63a
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a7cb90e63a | ||
|
|
1c598bb546 | ||
|
|
151af99765 | ||
|
|
75c004222c |
No files matched your search
@@ -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() {
|
||||||
|
|||||||
@@ -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:
|
||||||
|
# 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")
|
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
|
||||||
|
|
||||||
|
try:
|
||||||
import paho.mqtt.client as mqtt
|
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
|
||||||
|
|
||||||
|
try:
|
||||||
import paho.mqtt.client as mqtt
|
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 += [
|
||||||
|
|||||||
@@ -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`)는 새 옵션 거부 |
|
||||||
|
|||||||
@@ -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()
|
||||||
Reference in new issue
Block a user