toki/iop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(dispatcher): work log에 활성 파일과 loop를 기록한다

Browse source
This commit is contained in:
toki 2026-08-01 14:59:32 +09:00
parent aebba6508c
commit 4e85b24838
3 changed files with 180 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md
View file

@ -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.

90
agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py
View file

@ -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<loop>\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"(?<!\\)\|", stripped[1:-1])
]
return cells if len(cells) == 9 else None
if len(cells) == 10:
return cells
if len(cells) != 9:
return None
if cells[0] == "seq":
legacy_loop = "loop"
elif cells[0].startswith("---"):
legacy_loop = "---:"
else:
match = WORK_LOG_EXECUTION_LOOP_RE.search(cells[8])
legacy_loop = match.group("loop") if match else "0"
return [*cells[:4], legacy_loop, *cells[4:]]
def merge_work_log_sources(sources: set[Path]) -> 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"]),

116
agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py
View file

@ -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(
"<!-- task=m-principal-provider-credential-slot-routing/"
"02+01_credential_catalog plan=1 tag=TEST -->\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 <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,