fix(#34): 루프가 creator 의 하위 위임 잡까지 기다리고, 리뷰어에게 하위 잡 상태를 근거로 제공 #38

Merged
Godopu merged 9 commits from fix/issue-34 into main 2026-09-25 07:44:35 +00:00
3 changed files with 49 additions and 9 deletions
Showing only changes of commit 8a9101ed30 - Show all commits

No files matched your search

+11 -8
View File
@@ -19,7 +19,7 @@ import logging
import sys import sys
import uuid import uuid
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional, Tuple
from lib_py.delegation import mqtt_common from lib_py.delegation import mqtt_common
from lib_py.delegation.mqtt_common import ( from lib_py.delegation.mqtt_common import (
@@ -313,17 +313,20 @@ def _iter_records(registry_dir: str):
logger.warning("skipping unreadable record %s: %s", path, exc) logger.warning("skipping unreadable record %s: %s", path, exc)
def get_feedback(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> str: def get_feedback(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR,
"""Read the job's audit log or events log and return the detail of the last completed/error event.""" events: Tuple[str, ...] = ("completed", "error")) -> str:
"""Read the job's audit log or events log and return the detail of the last
event whose kind is in ``events`` (completed/error by default; Issue #34's
sub-job table adds ``cancelled`` to see why a job was never dispatched)."""
# 1) Try the unified audit log first (ndjson) since it's written synchronously by the subscriber # 1) Try the unified audit log first (ndjson) since it's written synchronously by the subscriber
try: try:
from lib_py.delegation import mqtt_common from lib_py.delegation import mqtt_common
logs_dir = mqtt_common.get_logs_dir() logs_dir = mqtt_common.get_logs_dir()
events = list(mqtt_common.iter_logged_events(job_id, logs_dir)) logged = list(mqtt_common.iter_logged_events(job_id, logs_dir))
for e in reversed(events): for e in reversed(logged):
if e.get("source_event") in ("completed", "error"): if e.get("source_event") in events:
return e.get("detail", "") return e.get("detail", "")
if e.get("event") in ("completed", "error"): if e.get("event") in events:
return e.get("detail", "") return e.get("detail", "")
except Exception: except Exception:
pass pass
@@ -339,7 +342,7 @@ def get_feedback(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> str:
continue continue
try: try:
payload = json.loads(line) payload = json.loads(line)
if payload.get("event") in ("completed", "error"): if payload.get("event") in events:
feedback = payload.get("detail", "") feedback = payload.get("detail", "")
except json.JSONDecodeError: except json.JSONDecodeError:
continue continue
+9 -1
View File
@@ -38,6 +38,8 @@ SUBJOB_RULE = ("규칙: 열린(OPEN)·미확인 종결(CLOSED-UNCONFIRMED)·실
"[VERDICT: NOT PASS] 를 내십시오. creator 리포트가 해당 부분의 대체 완료를 입증하면 예외입니다. " "[VERDICT: NOT PASS] 를 내십시오. creator 리포트가 해당 부분의 대체 완료를 입증하면 예외입니다. "
"CLOSED-UNCONFIRMED 는 오케스트레이터가 마감으로 닫은 잡이며 그 에이전트는 계속 일하고 있을 수 있습니다. " "CLOSED-UNCONFIRMED 는 오케스트레이터가 마감으로 닫은 잡이며 그 에이전트는 계속 일하고 있을 수 있습니다. "
"리뷰 중 소스가 바뀌면 이 회차의 승인은 무효가 됩니다.") "리뷰 중 소스가 바뀌면 이 회차의 승인은 무효가 됩니다.")
STRAY_RULE = ("규칙: 귀속되지 않은 열린 잡은 --parent-job 없이 등록된 이 작업의 하위 위임일 수 있습니다. 그 잡의 산출물이 "
"작업 목표에 필요하다면 산출물을 미완으로 보고 [VERDICT: NOT PASS] 를 내십시오.")
SELF_REVIEW_PROMPT = ("작업 완료 상태에 대해 스스로 검증(Self-Review)하여 결함이 없음을 확인하고 종결해주세요. " SELF_REVIEW_PROMPT = ("작업 완료 상태에 대해 스스로 검증(Self-Review)하여 결함이 없음을 확인하고 종결해주세요. "
"리뷰 리포트 마지막에 단독 행으로 반드시 '[VERDICT: PASS]' 혹은 '[VERDICT: NOT PASS]' 태그를 명시해주세요.") "리뷰 리포트 마지막에 단독 행으로 반드시 '[VERDICT: PASS]' 혹은 '[VERDICT: NOT PASS]' 태그를 명시해주세요.")
@@ -684,7 +686,9 @@ class LoopRunner:
status = str(r.get("status") or "?") status = str(r.get("status") or "?")
if status not in TERMINAL: if status not in TERMINAL:
return "OPEN", "-" return "OPEN", "-"
reason = job_registry.get_feedback(str(r.get("job_id")), str(self.jobs_dir)) or "" # merge review N1: a cancelled job's reason lives in its `cancelled` event
reason = job_registry.get_feedback(str(r.get("job_id")), str(self.jobs_dir),
events=("completed", "error", "cancelled")) or ""
if status == "completed": if status == "completed":
return "completed", reason or "-" return "completed", reason or "-"
if reason in ORCHESTRATOR_CLOSED or (status == "error" and not reason): if reason in ORCHESTRATOR_CLOSED or (status == "error" and not reason):
@@ -716,6 +720,10 @@ class LoopRunner:
stray_text = (" ".join(f"{r.get('job_id')}({r.get('agent_session', '?')}, {r.get('status')})" for r in stray) stray_text = (" ".join(f"{r.get('job_id')}({r.get('agent_session', '?')}, {r.get('status')})" for r in stray)
if stray else "없음") if stray else "없음")
stray_line = f"[귀속되지 않은 열린 잡] (부모 미표기, 이 작업 중 등록됨): {stray_text}\n" stray_line = f"[귀속되지 않은 열린 잡] (부모 미표기, 이 작업 중 등록됨): {stray_text}\n"
if stray:
# merge review K1: the case where --parent-job was left out is the one
# the table cannot show, so the rule has to name it explicitly
stray_line += STRAY_RULE + "\n"
if not rows and not stray: if not rows and not stray:
return "" # no sub-delegation: the prompts stay byte-identical (Issue #34 requirement 3) return "" # no sub-delegation: the prompts stay byte-identical (Issue #34 requirement 3)
if not rows: if not rows:
+29
View File
@@ -224,6 +224,33 @@ def test_t2b_the_wait_is_bounded_in_time_and_in_size(tmp_path, monkeypatch, caps
assert "over MAM_JOB_TREE_MAX=1" in capsys.readouterr().err assert "over MAM_JOB_TREE_MAX=1" in capsys.readouterr().err
def _cancel_before_injection(env: dict, reg: Path, jid: str, reason: str) -> None:
"""What the wrapper's finalize_unfinished does when the preflight cannot
see the target (target_not_found) or the session is busy."""
tok = _record(reg, jid)["dispatch_token"]
res = subprocess.run([sys.executable, str(REGISTRY), "--registry-dir", str(reg), "close", "--job", jid,
"--request", "cancelled", "--detail", reason, "--expect-dispatch-token", tok],
capture_output=True, text=True, env=env, timeout=120)
assert res.returncode == 0, res.stderr
_publish(env, reg, jid, "cancelled", reason)
@pytest.mark.parametrize("reason", ["target_not_found", "session_busy"])
def test_t13_a_sub_job_cancelled_before_injection_is_not_dispatched_in_the_table(tmp_path, monkeypatch, reason):
"""Merge review N1: get_feedback ignored `cancelled` events, so the table
showed a bare `cancelled` and the reason (never dispatched vs. the agent
failing) was lost."""
env = _env(tmp_path)
reg = tmp_path / "reg"
p = _jid(_submit(env, reg))
c = _jid(_submit(env, reg, "--parent-job", p))
_cancel_before_injection(env, reg, c, reason)
assert _record(reg, c)["status"] == "cancelled"
block = _runner(tmp_path, monkeypatch)._subjobs_block([p])
assert f"| {c} | " in block, block
assert f"| NOT-DISPATCHED | {reason} |" in block, block
def test_t7_a_job_without_children_is_waited_for_as_before(tmp_path, monkeypatch): def test_t7_a_job_without_children_is_waited_for_as_before(tmp_path, monkeypatch):
env = _env(tmp_path) env = _env(tmp_path)
reg = tmp_path / "reg" reg = tmp_path / "reg"
@@ -458,6 +485,8 @@ def test_t6_an_unattributed_open_job_is_shown_and_waited_for_only_on_request(tmp
prompt = _review_prompt(ctl) prompt = _review_prompt(ctl)
assert "[하위 위임 잡] 없음" in prompt assert "[하위 위임 잡] 없음" in prompt
assert re.search(rf"\[귀속되지 않은 열린 잡\] .*: {uid}\(herdr:img, pending\)", prompt), prompt assert re.search(rf"\[귀속되지 않은 열린 잡\] .*: {uid}\(herdr:img, pending\)", prompt), prompt
# merge review K1: the reviewer gets a rule for that job, not only the fact
assert "귀속되지 않은 열린 잡은 --parent-job 없이" in prompt and "[VERDICT: NOT PASS]" in prompt
ctl2 = _loop_ctl(tmp_path / "strict") ctl2 = _loop_ctl(tmp_path / "strict")
res2 = _run_loop(ctl2, _loop_env(ctl2, FAKE_WORKER="stray", MAM_CHILD_WAIT_S="3"), res2 = _run_loop(ctl2, _loop_env(ctl2, FAKE_WORKER="stray", MAM_CHILD_WAIT_S="3"),