feat(dispatcher): 작업 파일 대상을 로그에 표시한다

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

@@ -482,6 +482,24 @@ class Task:
     grade: int | None = None
 
 
+def task_target_files(task: Task) -> list[str]:
+    """Return the canonical plan-declared file targets for dispatcher output."""
+    return sorted(str(Path(path).resolve()) for path in task.write_set)
+
+
+def task_observation_lines(task: Task) -> list[str]:
+    """Render the task directory and declared file targets for operator logs."""
+    lines = [f"task_dir={task.directory.resolve()}"]
+    targets = task_target_files(task)
+    if targets:
+        lines.extend(f"target_file={path}" for path in targets)
+    else:
+        lines.append(
+            "target_file=unavailable (Modified Files Summary has no valid file claim)"
+        )
+    return lines
+
+
 def next_execution_identity(
     store: StateStore,
     task: Task,
@@ -3701,6 +3719,9 @@ async def invoke(
     record: dict[str, Any] = {
         "execution_id": identity,
         "task": task.name,
+        "task_directory": str(task.directory.resolve()),
+        "target_files": task_target_files(task),
+        "target_files_known": task.write_set_known,
         "plan_number": plan_number(task),
         "role": role,
         "attempt": attempt,
@@ -3777,6 +3798,8 @@ async def invoke(
                 f"locator 기록 경고: locator={locator_path} error={exc}",
             )
 
+    for line in task_observation_lines(task):
+        attempt_event(prefix, line)
     attempt_event(prefix, f"locator={locator_path}")
     try:
         append_milestone_event(
@@ -5510,6 +5533,7 @@ async def run_worker(
             f"model={spec.display}",
             f"plan={task.plan.resolve()}",
             f"work_log={work_log.resolve()}",
+            *task_observation_lines(task),
         ],
     )
     success, locator = await run_escalating(
@@ -5641,6 +5665,7 @@ async def run_selfcheck(
             f"model={spec.display}",
             f"plan={task.plan.resolve()}",
             f"work_log={work_log.resolve()}",
+            *task_observation_lines(task),
         ],
     )
     # 0 means the full pass is pending. After it fails the checklist gate,
@@ -5801,7 +5826,15 @@ async def run_review(
     before = task_signature(workspace, task)
     prior_review_fingerprints = review_fingerprints(workspace, task)
     target = task.review.resolve() if task.review else task.directory.resolve()
-    banner("리뷰시작", task.name, [f"model={spec.display}", f"review={target}"])
+    banner(
+        "리뷰시작",
+        task.name,
+        [
+            f"model={spec.display}",
+            f"review={target}",
+            *task_observation_lines(task),
+        ],
+    )
     success, locator = await run_escalating(
         workspace,
         store,

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

@@ -1974,6 +1974,43 @@ class WorkLogInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase):
             self.assertIn("| START | test | worker | 0 | pi | running |", log)
             self.assertIn("| FINISH | test | worker | 0 | pi | succeeded:0 |", log)
 
+    async def test_invoke_logs_task_directory_and_plan_declared_file_targets(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            workspace = Path(temporary)
+            (workspace / ".git").mkdir()
+            task = TaskStageTest().make_task(workspace)
+            store = dispatch.StateStore(workspace)
+            spec = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True)
+            command = [sys.executable, "-c", "print('done')"]
+            try:
+                with (
+                    mock.patch.object(
+                        dispatch, "build_command", return_value=command
+                    ),
+                    mock.patch("sys.stdout", new_callable=io.StringIO) as stdout,
+                ):
+                    rc, failure, locator = await dispatch.invoke(
+                        workspace,
+                        store,
+                        task,
+                        "worker",
+                        spec,
+                        "Read the plan.",
+                    )
+            finally:
+                store.close()
+
+            self.assertEqual(rc, 0)
+            self.assertIsNone(failure)
+            target = str((workspace / "src" / "test.py").resolve())
+            output = stdout.getvalue()
+            self.assertIn(f"task_dir={workspace.resolve()}", output)
+            self.assertIn(f"target_file={target}", output)
+            record = json.loads(locator.read_text(encoding="utf-8"))
+            self.assertEqual(record["task_directory"], str(workspace.resolve()))
+            self.assertEqual(record["target_files"], [target])
+            self.assertTrue(record["target_files_known"])
+
     async def test_invoke_does_not_require_model_written_work_log(self):
         with tempfile.TemporaryDirectory() as temporary:
             workspace = Path(temporary)