fix(issue-34): merge review - a sub-job cancelled before injection keeps its reason in the review table; unattributed open jobs come with a verdict rule
- registry.get_feedback only returned the detail of the last completed or error event, so a sub-job the wrapper cancelled before injection (target_not_found, session_busy) reached the review table as a bare `cancelled` and its reason was lost (merge review, grok N1). get_feedback takes the event kinds to read (default unchanged); the table asks for cancelled too and shows NOT-DISPATCHED with the reason. Test T13, both reasons, fails without the fix. - the [귀속되지 않은 열린 잡] line listed jobs registered without --parent-job but gave the reviewer no rule for them - the one case the table cannot show (claude K1). A rule sentence follows the line whenever such a job exists; without one the prompt is unchanged. T6 pins it. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff
This commit is contained in:
1 parent
ed293dcef4
commit
8a9101ed30
3 files changed
+49
-9
No files matched your search
@@ -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
|
||||||
|
|||||||
@@ -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:
|
||||||
|
|||||||
@@ -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"),
|
||||||
|
|||||||
Reference in new issue
Block a user