fix(mqtt): resolve B-9 by implementing lazy get_logs_dir() evaluation

- Replace import-time LOGS_DIR cwd binding with dynamic get_logs_dir() function
- Implement PEP 562 __getattr__ and __dir__ for transparent LOGS_DIR backward compatibility
- Update audit-log callers in mqtt_common.py and registry.py to use dynamic resolution
- Add 5 regression guards in tests/test_tier1_unit.py (276/276 PASS)
- Update IMPROVEMENTS.md, VERSIONS.md, registry.md, and include plan and peer review reports
This commit is contained in:
2026-08-17 13:37:51 +09:00
parent 8cee9374b1
commit ac82f9b993
8 changed files with 671 additions and 22 deletions
+71
View File
@@ -402,3 +402,74 @@ def test_b10_find_workspace_uuid_runs_without_pyyaml(tmp_path):
assert r.returncode == 0, f"resolution path still needs PyYAML: {r.stderr}"
assert "yaml" not in r.stderr.lower(), f"PyYAML touched at runtime: {r.stderr}"
# ===========================================================================
# B-9: LOGS_DIR import-time cwd binding resolution regression guards
# ===========================================================================
def test_b9_logs_dir_follows_cwd_changes(mam_sandbox, tmp_path, monkeypatch):
"""B-9: the audit-log root must be resolved per call, not frozen at import."""
mq = get_mqtt_common(mam_sandbox)
monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR", raising=False)
a = tmp_path / "a"; b = tmp_path / "b"
a.mkdir(); b.mkdir()
# realpath on both sides: pytest's tmp_path happens to be pre-resolved today,
# but relying on that is an undocumented dependency (C1').
def logs_under(p):
return os.path.realpath(os.path.join(str(p), ".mam", "delegate_job_logs"))
monkeypatch.chdir(a)
assert os.path.realpath(mq.get_logs_dir()) == logs_under(a)
monkeypatch.chdir(b)
assert os.path.realpath(mq.get_logs_dir()) == logs_under(b)
# the compat alias must follow too (T1: a surviving global fails here)
assert os.path.realpath(mq.LOGS_DIR) == logs_under(b)
def test_b9_audit_log_lands_under_the_current_cwd(mam_sandbox, tmp_path, monkeypatch):
"""B-9/T3: assert the FILE appears — a swallowed NameError must not pass."""
mq = get_mqtt_common(mam_sandbox)
monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR", raising=False)
monkeypatch.chdir(tmp_path)
mq.init_job_log("b9job", {"status": "pending"})
assert (tmp_path / ".mam" / "delegate_job_logs" / "b9job" / "meta.json").exists(), \
"audit log did not land under the current cwd (the best-effort handler may have swallowed an error)"
def test_b9_logs_dir_env_override_is_dynamic(mam_sandbox, tmp_path, monkeypatch):
"""B-9: DELEGATE_JOB_LOGS_DIR must be honoured at call time, both ways."""
mq = get_mqtt_common(mam_sandbox)
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", "/tmp/b9-override")
assert mq.get_logs_dir() == "/tmp/b9-override"
monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR")
# equality, not inequality — clearing the env must restore the cwd default (Rev.2 §3)
assert os.path.realpath(mq.get_logs_dir()) == \
os.path.realpath(os.path.join(str(tmp_path), ".mam", "delegate_job_logs"))
def test_b9_no_module_level_logs_dir_binding():
"""B-9/T1: a surviving module global would make __getattr__ dead code."""
import ast, pathlib
src = (pathlib.Path(__file__).resolve().parent.parent / ".agents" / "skills"
/ "multi-agent-mux-delegate-job" / "scripts" / "mqtt_common.py")
tree = ast.parse(src.read_text())
for node in tree.body: # module scope only
if isinstance(node, ast.Assign):
for t in node.targets:
assert not (isinstance(t, ast.Name) and t.id == "LOGS_DIR"), \
f"line {node.lineno}: module-level LOGS_DIR binding shadows __getattr__ (B-9/T1)"
elif isinstance(node, ast.AnnAssign): # C2-b: LOGS_DIR: str = ... parses as AnnAssign
assert not (isinstance(node.target, ast.Name) and node.target.id == "LOGS_DIR"), \
f"line {node.lineno}: annotated module-level LOGS_DIR binding shadows __getattr__ (B-9/T1)"
def test_b9_logs_dir_stays_discoverable(mam_sandbox):
"""B-9/C2-a: PEP 562 __dir__ keeps LOGS_DIR visible to dir() and tooling."""
mq = get_mqtt_common(mam_sandbox)
assert "LOGS_DIR" in dir(mq)
assert hasattr(mq, "LOGS_DIR") # true via __getattr__ even without __dir__
assert dir(mq).count("LOGS_DIR") == 1 # set-based __dir__ must not duplicate