From 4e85b248381c97836b13aad10ed059c9dbf5961f Mon Sep 17 00:00:00 2001 From: toki Date: Sat, 1 Aug 2026 14:59:32 +0900 Subject: [PATCH] =?UTF-8?q?fix(dispatcher):=20work=20log=EC=97=90=20?= =?UTF-8?q?=ED=99=9C=EC=84=B1=20=ED=8C=8C=EC=9D=BC=EA=B3=BC=20loop?= =?UTF-8?q?=EB=A5=BC=20=EA=B8=B0=EB=A1=9D=ED=95=9C=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../orchestrate-agent-task-loop/SKILL.md | 2 +- .../scripts/dispatch.py | 90 ++++++++++---- .../tests/test_dispatch.py | 116 +++++++++++++++++- 3 files changed, 180 insertions(+), 28 deletions(-) diff --git a/agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md b/agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md index d82f902..191d2e4 100644 --- a/agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md +++ b/agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md @@ -123,7 +123,7 @@ When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a termin - Keep exactly one `agent-task/{task_group}/WORK_LOG.md` per task group. Do not create one in a split-subtask directory. - Allow only the dispatcher to modify this file. Worker/self-check/review models need not read or update it, and success must not depend on its prose. -- Append chronological `START`/`FINISH` rows with time, task, role, attempt, model, result, and locator. Record time in KST (`UTC+09:00`) as `YY-MM-DD HH:MM:SS`, for example `26-07-26 07:40:15`. Use this single timeline to inspect parallel execution order. +- Append chronological `START`/`FINISH` rows with time, task, loop, role, attempt, model, result, and locator. In `task`, record the active role artifact relative to `agent-task/`: the PLAN path for a worker and the CODE_REVIEW path for self-check/review. In `loop`, record the PLAN identity's zero-based `plan` number (`0` is the initial plan). Record time in KST (`UTC+09:00`) as `YY-MM-DD HH:MM:SS`, for example `26-07-26 07:40:15`. Use this single timeline to inspect parallel execution order. - Do not require the common code-review skill to preserve `WORK_LOG.md`. For split work the group log normally remains in the parent because review moves only the selected subtask. For a single task review may move the log with the task archive; after review exits, resolve exactly one source from the active group path or verified completed archive and normalize it to `work_log_N.log`. - After every observed task in a task group has a verified complete archive and no active/running task remains, append the final `FINISH` and move the generated `WORK_LOG.md` under the final completed archive's group root as `work_log_N.log`. If an archive exists after restart but the last `START` lacks `FINISH`, do not terminate or archive while any PID/start token, per-attempt process marker, or pidless stream/native evidence remains live. Track it until execution evidence has ended and the complete archive is verified, then append `FINISH` with `reconciled:verified-complete-archive` and move the log. Use `agent-task/archive/YYYY/MM/{task_group}/` for split tasks and the actual suffix-bearing archive destination for a single task. Set `N` to one more than the maximum suffix for the same task group across all months, starting at `0`. - If `WORK_LOG.md` archiving fails or multiple active/archive sources exist, drain other independent work and return non-terminal exit `3` for retry. Return successful exit `0` only after a completed group that generated a log has no active `WORK_LOG.md` and its `work_log_N.log` is verified. Keep an incomplete group's `WORK_LOG.md` active for blocker or exit `3` recovery. diff --git a/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py b/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py index 57114bc..54e0e20 100644 --- a/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py +++ b/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py @@ -118,6 +118,17 @@ IMPLEMENTATION_CHECKBOX_RE = re.compile( ) WORK_LOG_NAME = "WORK_LOG.md" WORK_LOG_ARCHIVE_RE = re.compile(r"^work_log_(\d+)\.log$") +WORK_LOG_HEADER = ( + "| seq | time | event | task | loop | role | attempt | model | result | locator |" +) +WORK_LOG_SEPARATOR = "|---:|---|---|---|---:|---|---:|---|---|---|" +LEGACY_WORK_LOG_HEADER = ( + "| seq | time | event | task | role | attempt | model | result | locator |" +) +LEGACY_WORK_LOG_SEPARATOR = "|---:|---|---|---|---|---:|---|---|---|" +WORK_LOG_EXECUTION_LOOP_RE = re.compile( + r"__p(?P\d+)__(?:worker|selfcheck|review)__a\d+(?=$|[/\\])" +) AGENT_PROCESS_MARKER_ENV = "IOP_AGENT_TASK_EXECUTION_ID" DISPATCHER_CHILD_BOUNDARY_PROMPT = ( "You are a child agent already launched by the dispatcher, not the " @@ -286,10 +297,25 @@ def milestone_work_log_path(task: Task) -> Path: ) +def work_log_task_name(task: Task, role: str) -> str: + """Return the role-specific active artifact shown in the task column.""" + artifact = task.plan if role == "worker" else task.review + if artifact is None: + return task.name + return f"{task.name}/{artifact.name}" + + +def work_log_loop_number(task: Task, execution_id: str) -> int: + """Keep one loop identity even when a reviewer archives the active PLAN.""" + match = WORK_LOG_EXECUTION_LOOP_RE.search(execution_id) + return int(match.group("loop")) if match else plan_number(task) + + def append_work_log_event( path: Path, *, task_name: str, + loop: int, event: str, execution_id: str, role: str, @@ -308,20 +334,20 @@ def append_work_log_event( stream.write( "# Milestone Work Log\n\n" "> Dispatcher-owned execution timeline. Workers and reviewers do not edit this file.\n\n" - "| seq | time | event | task | role | attempt | model | result | locator |\n" - "|---:|---|---|---|---|---:|---|---|---|\n" + f"{WORK_LOG_HEADER}\n" + f"{WORK_LOG_SEPARATOR}\n" ) sequence = 1 else: stream.seek(0, os.SEEK_END) - if "| seq | time | event | task | role | attempt | model | result | locator |" not in text: + if WORK_LOG_HEADER not in text: if not text.endswith("\n"): stream.write("\n") stream.write( "\n## Dispatcher Timeline\n\n" "> Dispatcher-owned. Workers and reviewers do not edit this section.\n\n" - "| seq | time | event | task | role | attempt | model | result | locator |\n" - "|---:|---|---|---|---|---:|---|---|---|\n" + f"{WORK_LOG_HEADER}\n" + f"{WORK_LOG_SEPARATOR}\n" ) sequence = 1 + max( ( @@ -339,7 +365,7 @@ def append_work_log_event( stream.write( f"| {sequence} | {work_log_now_kst()} | {cell(event)} | " f"{cell(task_name)} | " - f"{cell(role)} | {attempt} | {cell(model)} | {cell(result)} | " + f"{loop} | {cell(role)} | {attempt} | {cell(model)} | {cell(result)} | " f"{cell(locator.resolve())} |\n" ) stream.flush() @@ -361,7 +387,8 @@ def append_milestone_event( ) -> Path: return append_work_log_event( milestone_work_log_path(task), - task_name=task.name, + task_name=work_log_task_name(task, role), + loop=work_log_loop_number(task, execution_id), event=event, execution_id=execution_id, role=role, @@ -5056,7 +5083,18 @@ def work_log_event_cells(line: str) -> list[str] | None: cell.strip().replace(r"\|", "|") for cell in re.split(r"(? str: @@ -5066,10 +5104,12 @@ def merge_work_log_sources(sources: set[Path]) -> str: "## Dispatcher Timeline", "> Dispatcher-owned execution timeline. Workers and reviewers do not edit this file.", "> Dispatcher-owned. Workers and reviewers do not edit this section.", - "| seq | time | event | task | role | attempt | model | result | locator |", - "|---:|---|---|---|---|---:|---|---|---|", + WORK_LOG_HEADER, + WORK_LOG_SEPARATOR, + LEGACY_WORK_LOG_HEADER, + LEGACY_WORK_LOG_SEPARATOR, } - unique_rows: dict[tuple[str, str, str, str, str], tuple[str, ...]] = {} + unique_rows: dict[tuple[str, ...], tuple[str, ...]] = {} ordered_rows: list[tuple[str, int, int, list[str]]] = [] for source_index, source in enumerate(sorted(sources)): @@ -5100,21 +5140,22 @@ def merge_work_log_sources(sources: set[Path]) -> str: ) try: sequence = int(cells[0]) - int(cells[5]) + int(cells[4]) + int(cells[6]) except ValueError as exc: raise ValueError( - "WORK_LOG 병합 대상의 sequence 또는 attempt가 유효하지 않다: " + "WORK_LOG 병합 대상의 sequence, loop 또는 attempt가 유효하지 않다: " f"source={source} line={line_number}" ) from exc - key = (cells[2], cells[3], cells[4], cells[5], cells[8]) + key = (cells[2], cells[3], cells[4], cells[5], cells[6], cells[9]) fingerprint = tuple(cells[1:]) previous = unique_rows.get(key) if previous is not None: if previous != fingerprint: raise ValueError( "WORK_LOG 병합 충돌: " - f"event={cells[2]} task={cells[3]} role={cells[4]} " - f"attempt={cells[5]} locator={cells[8]}" + f"event={cells[2]} task={cells[3]} loop={cells[4]} " + f"role={cells[5]} attempt={cells[6]} locator={cells[9]}" ) continue unique_rows[key] = fingerprint @@ -5126,8 +5167,8 @@ def merge_work_log_sources(sources: set[Path]) -> str: header = ( "# Milestone Work Log\n\n" "> Dispatcher-owned execution timeline. Workers and reviewers do not edit this file.\n\n" - "| seq | time | event | task | role | attempt | model | result | locator |\n" - "|---:|---|---|---|---|---:|---|---|---|\n" + f"{WORK_LOG_HEADER}\n" + f"{WORK_LOG_SEPARATOR}\n" ) rendered_rows: list[str] = [] for sequence, (_, _, _, cells) in enumerate(sorted(ordered_rows), start=1): @@ -5153,20 +5194,22 @@ def unfinished_work_log_attempts(path: Path) -> list[dict[str, Any]]: continue try: sequence = int(cells[0]) - attempt = int(cells[5]) + loop = int(cells[4]) + attempt = int(cells[6]) except ValueError: continue - locator = cells[8] + locator = cells[9] key = locator or "\0".join( - (cells[3], cells[4], cells[5], cells[6]) + (cells[3], cells[4], cells[5], cells[6], cells[7]) ) if cells[2] == "START": open_attempts[key] = { "sequence": sequence, "task_name": cells[3], - "role": cells[4], + "loop": loop, + "role": cells[5], "attempt": attempt, - "model": cells[6], + "model": cells[7], "locator": locator, } else: @@ -5185,6 +5228,7 @@ def close_unfinished_work_log_attempts(path: Path) -> int: append_work_log_event( path, task_name=str(record["task_name"]), + loop=int(record["loop"]), event="FINISH", execution_id=f"reconciled-{record['sequence']}", role=str(record["role"]), diff --git a/agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py b/agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py index 649012c..b66cfa4 100644 --- a/agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py +++ b/agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py @@ -1916,6 +1916,99 @@ class WorkLogInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase): self.assertRegex(dispatch.now_iso(), r"\+00:00$") + def test_milestone_timeline_uses_active_artifact_and_plan_loop(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + task_directory = ( + workspace + / "agent-task" + / "m-principal-provider-credential-slot-routing" + / "02+01_credential_catalog" + ) + task_directory.mkdir(parents=True) + plan = task_directory / "PLAN-local-G07.md" + review = task_directory / "CODE_REVIEW-cloud-G07.md" + plan.write_text( + "\n", + encoding="utf-8", + ) + review.write_text("review\n", encoding="utf-8") + task = dispatch.Task( + name=( + "m-principal-provider-credential-slot-routing/" + "02+01_credential_catalog" + ), + directory=task_directory, + plan=plan, + review=review, + user_review=None, + recovery=False, + ) + + for role in ("worker", "selfcheck", "review"): + dispatch.append_milestone_event( + task, + event="START", + execution_id=f"test__p1__{role}__a00", + role=role, + attempt=0, + model="test-model", + result="running", + locator=workspace / role / "locator.json", + ) + + plan.rename(task_directory / "plan_local_G07_1.log") + dispatch.append_milestone_event( + task, + event="FINISH", + execution_id="test__p1__review__a00", + role="review", + attempt=0, + model="test-model", + result="succeeded:0", + locator=workspace / "review" / "locator.json", + ) + + log = ( + task_directory.parent / dispatch.WORK_LOG_NAME + ).read_text(encoding="utf-8") + self.assertIn( + "| seq | time | event | task | loop | role | attempt |", + log, + ) + task_name = task.name + self.assertIn( + f"| START | {task_name}/PLAN-local-G07.md | 1 | worker | 0 |", + log, + ) + self.assertIn( + f"| START | {task_name}/CODE_REVIEW-cloud-G07.md | " + "1 | selfcheck | 0 |", + log, + ) + self.assertIn( + f"| START | {task_name}/CODE_REVIEW-cloud-G07.md | " + "1 | review | 0 |", + log, + ) + self.assertIn( + f"| FINISH | {task_name}/CODE_REVIEW-cloud-G07.md | " + "1 | review | 0 |", + log, + ) + + def test_legacy_timeline_infers_loop_from_locator_identity(self): + cells = dispatch.work_log_event_cells( + "| 21 | 26-08-01 14:18:01 | START | group/task | selfcheck | " + "0 | pi | running | /workspace__p9__/runs/" + "group__task__p1__selfcheck__a00/locator.json |" + ) + + self.assertIsNotNone(cells) + assert cells is not None + self.assertEqual(cells[3:7], ["group/task", "1", "selfcheck", "0"]) + async def test_invoke_writes_dispatcher_owned_milestone_timeline(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) @@ -1971,8 +2064,15 @@ class WorkLogInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase): self.assertIn("Dispatcher-owned execution timeline", log) self.assertRegex(log, r"\| \d+ \| \d{2}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} \| START \|") self.assertRegex(log, r"\| \d+ \| \d{2}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} \| FINISH \|") - self.assertIn("| START | test | worker | 0 | pi | running |", log) - self.assertIn("| FINISH | test | worker | 0 | pi | succeeded:0 |", log) + self.assertIn( + "| START | test/PLAN-local-G05.md | 0 | worker | 0 | pi | running |", + log, + ) + self.assertIn( + "| FINISH | test/PLAN-local-G05.md | 0 | worker | 0 | pi | " + "succeeded:0 |", + log, + ) async def test_invoke_logs_task_directory_and_plan_declared_file_targets(self): with tempfile.TemporaryDirectory() as temporary: @@ -2232,7 +2332,11 @@ class WorkLogInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(record["exit_code"], "cancelled") self.assertEqual(record["failure_class"], "cancelled") log = (workspace / dispatch.WORK_LOG_NAME).read_text(encoding="utf-8") - self.assertIn("| FINISH | test | worker | 0 | pi | failed:cancelled |", log) + self.assertIn( + "| FINISH | test/PLAN-local-G05.md | 0 | worker | 0 | pi | " + "failed:cancelled |", + log, + ) async def test_pi_silent_awaiting_model_is_inspected_without_termination(self): with tempfile.TemporaryDirectory() as temporary: @@ -5933,6 +6037,7 @@ class WorkLogArchiveTest(unittest.TestCase): dispatch.append_work_log_event( source, task_name="group/01_done", + loop=0, event="START", execution_id="group__01_done__p0__review__a00", role="review", @@ -11811,7 +11916,10 @@ class ArtifactLanguageContractTest(unittest.TestCase): self.assertIn("dispatch.py --workspace --validate-plan", document) self.assertIn("exact workspace", document) self.assertIn("Never use a glob (`*`, `?`, `[]`)", plan_skill) - self.assertIn("globs, directories, workspace root", review_skill) + self.assertIn( + "globs, directories, workspace root", + review_skill.casefold(), + ) self.assertIn( "Fail the task closed when any path is broad", orchestrator_skill,