From 211d33e1f525a917dc11674d2a2125f991d0bc47 Mon Sep 17 00:00:00 2001 From: toki Date: Wed, 29 Jul 2026 08:07:43 +0900 Subject: [PATCH] =?UTF-8?q?feat(agent-ops):=20=EC=9E=91=EC=97=85=EA=B3=B5?= =?UTF-8?q?=EA=B0=84=20=EC=86=8C=EC=9C=A0=EA=B6=8C=20=EA=B2=80=EC=A6=9D?= =?UTF-8?q?=EC=9D=84=20=EC=B6=94=EA=B0=80=ED=95=9C=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 같은 물리 작업공간에서 충돌하는 병렬 작업과 외부 실행 상태가 서로 영향을 주지 않도록 한다. --- .../agent-ui/sync-state-template.json | 3 +- .../skills/common/sync-agent-ui/SKILL.md | 112 ++- agent-ops/skills/common/update-test/SKILL.md | 84 +- .../orchestrate-agent-task-loop/SKILL.md | 34 +- .../scripts/dispatch.py | 529 ++++++++++- .../tests/test_dispatch.py | 865 ++++++++++++++++-- .../code_review_cloud_G08_0.log | 183 ++++ .../complete.log | 34 + .../plan_cloud_G08_0.log | 219 +++++ 9 files changed, 1929 insertions(+), 134 deletions(-) create mode 100644 agent-task/archive/2026/07/dispatcher_workspace_ownership/code_review_cloud_G08_0.log create mode 100644 agent-task/archive/2026/07/dispatcher_workspace_ownership/complete.log create mode 100644 agent-task/archive/2026/07/dispatcher_workspace_ownership/plan_cloud_G08_0.log diff --git a/agent-ops/skills/common/_templates/agent-ui/sync-state-template.json b/agent-ops/skills/common/_templates/agent-ui/sync-state-template.json index 6b1ba1c..f39c315 100644 --- a/agent-ops/skills/common/_templates/agent-ui/sync-state-template.json +++ b/agent-ops/skills/common/_templates/agent-ui/sync-state-template.json @@ -1,5 +1,5 @@ { - "schema_version": 1, + "schema_version": 2, "surface_type": "ops-dev", "baseline_mode": "", "last_sync_mode": "baseline", @@ -7,5 +7,6 @@ "last_synced_at": "", "agent_ui_paths": [], "code_paths": [], + "pending_code_work": [], "notes": [] } diff --git a/agent-ops/skills/common/sync-agent-ui/SKILL.md b/agent-ops/skills/common/sync-agent-ui/SKILL.md index 9f45986..fb14bf3 100644 --- a/agent-ops/skills/common/sync-agent-ui/SKILL.md +++ b/agent-ops/skills/common/sync-agent-ui/SKILL.md @@ -1,7 +1,7 @@ --- name: sync-agent-ui -version: 1.1.0 -description: 정합화된 agent-ui 변경분을 코드에 직접 반영하거나, 코드 작업 규모에 따라 plan 또는 roadmap/milestone 작업으로 라우팅한다. 직접 반영이 검증된 경우에만 .sync-state.json 기준점을 갱신한다. "agent-ui와 코드 동기화", "화면정의서대로 코드 반영" 요청 시 사용한다. +version: 1.2.0 +description: 정합화된 agent-ui 변경분을 코드에 직접 반영하거나 plan/roadmap으로 라우팅하고, plan-required 작업 매핑과 완료 후 agent-ui 상태 정합화까지 소유하는 스킬. "agent-ui와 코드 동기화", "화면정의서대로 코드 반영", agent-ui 코드 작업 완료 정합화 요청 시 사용한다. --- # sync-agent-ui @@ -11,38 +11,105 @@ description: 정합화된 agent-ui 변경분을 코드에 직접 반영하거나 정합화된 `agent-ui/` 변경분을 실제 UI 코드에 반영한다. 기본 동작은 `.sync-state.json`의 `last_synced_head` 이후 agent-ui 변경분만 반영하며, 사용자가 전체 동기화를 명시한 경우에만 현재 agent-ui 전체를 코드와 대조한다. 코드 반영 전에 코드 작업 규모를 판정해 작은 작업은 직접 반영하고, 중간 작업은 `plan` 루프로, Milestone이 필요한 큰 코드 작업은 `update-roadmap`의 Milestone/Epic/Task 갱신으로 넘긴다. -`plan-required` 또는 `milestone-required`로 라우팅한 경우에는 이 스킬 실행 안에서 코드 구현, status 전환, sync state 갱신, commit/push를 하지 않는다. +`plan-required` 작업은 일반 plan/code-review 문서에 agent-ui 전용 section을 넣지 않는다. 대신 plan pair 생성 직후 `prepare-code-work`로 task와 agent-ui 문서를 매핑하고, 일반 code-review PASS와 `complete.log` 생성 뒤 `reconcile-completion`으로 status/code evidence를 정합화한다. +`milestone-required`로 라우팅한 경우에는 이 스킬 실행 안에서 코드 구현, status 전환, sync state 갱신, commit/push를 하지 않는다. ## 언제 호출할지 - 사용자가 "agent-ui와 코드 동기화", "화면정의서대로 코드 반영", "wireframe 변경사항 구현"을 요청할 때 - `update-agent-ui` 또는 `validate-agent-ui`가 code sync 필요성을 판단했고 현재 범위에 미해결 USER_REVIEW가 없을 때 - 사용자가 "agent-ui와 코드 전체 동기화"처럼 full sync를 명시할 때 +- `plan-required`로 생성된 task와 agent-ui 문서의 연결을 기록할 때 +- 일반 code-review PASS 뒤 생성된 `complete.log`를 agent-ui status/code evidence에 반영할 때 ## 입력 -- `mode`: `incremental` 또는 `full`. 기본값은 `incremental` (선택) +- `mode`: `incremental`, `full`, `prepare-code-work`, `reconcile-completion` 중 하나. 기본값은 `incremental` (선택) - `scope`: `all`, `view:`, `component:` 중 하나. 기본값은 `all` (선택) - `validated`: `true` 또는 `false`. 기본값은 `false` (선택) - `sync-intent-source`: `manual`, `update-agent-ui`, `validate-agent-ui` 중 하나 (선택) - `execution-route`: `auto`, `direct-sync`, `plan-required`, `milestone-required` 중 하나. 기본값은 `auto` (선택) +- `task-path`: `agent-task/` 기준 원래 active task 경로. `prepare-code-work`에서 필수, `reconcile-completion`에서 선택 +- `agent-ui-docs`: 구현 대상 활성 definition/frame/component 문서 목록. `prepare-code-work`에서 필수 +- `code-paths`: 구현 후보 또는 완료 evidence 코드 경로 목록. `prepare-code-work`에서 필수 +- `verification-requirements`: 완료 정합화 전에 확인할 명령/기대 결과 목록. `prepare-code-work`에서 선택 +- `completion-log`: 일반 code-review가 만든 정확한 `complete.log` 경로. `reconcile-completion`에서 필수 +- `sync-result-head`: agent-ui와 코드 결과를 모두 포함한다고 검증된 commit hash. `reconcile-completion`에서 선택 - `push`: `true` 또는 `false`. 기본값은 `true` (선택) ## 먼저 확인할 것 - [ ] `agent-ops/rules/common/rules-agent-ui.md`를 읽는다. - [ ] `agent-ui/`와 `agent-ui/definition/index.md` 존재 여부를 확인한다. -- [ ] `validated=true`가 아니면 먼저 `validate-agent-ui`를 실행한다. +- [ ] `mode=incremental|full|prepare-code-work`이고 `validated=true`가 아니면 먼저 `validate-agent-ui`를 실행한다. - [ ] `agent-ui/USER_REVIEW.md`가 있으면 현재 sync 범위와 관련된 미해결 항목이 없는지 확인한다. -- [ ] `agent-ui/.sync-state.json`을 읽는다. 없고 `mode=incremental`이면 legacy baseline migration 필요로 보고 code sync를 시작하지 않는다. -- [ ] 수정할 UI 코드 경로에 해당하는 project/domain rule을 먼저 읽는다. -- [ ] 코드 변경 검증이 필요하면 작업 환경의 `agent-test//rules.md`를 읽는다. 환경 미지정은 `local`로 본다. +- [ ] `agent-ui/.sync-state.json`을 읽는다. 없고 `mode=incremental|prepare-code-work|reconcile-completion`이면 legacy baseline migration 필요로 보고 중단한다. +- [ ] `schema_version: 1`이고 `pending_code_work`가 없으면 빈 배열로 취급해 호환하고, 이 스킬이 state를 실제 갱신할 때만 `schema_version: 2`로 올린다. +- [ ] `mode=incremental|full`이면 수정할 UI 코드 경로에 해당하는 project/domain rule을 먼저 읽는다. +- [ ] `mode=incremental|full`이고 코드 변경 검증이 필요하면 작업 환경의 `agent-test//rules.md`를 읽는다. 환경 미지정은 `local`로 본다. +- [ ] `mode=prepare-code-work`이면 exact active `PLAN-*-G??.md`와 matching `CODE_REVIEW-*-G??.md`가 `task-path`에 존재하는지 확인한다. +- [ ] `mode=reconcile-completion`이면 exact `completion-log`가 존재하고 matching pending entry가 정확히 하나인지 확인한다. - [ ] `git status --short`와 현재 `HEAD`를 확인한다. - [ ] unrelated dirty worktree가 있으면 sync commit 전에 범위를 보고하고, agent-ui와 관련 코드 파일만 stage한다. - [ ] 이전 sync 실패로 남은 관련 작업트리 변경이 있으면 보존할지, 사용자 결정과 충돌하는지 확인한다. ## 실행 절차 +### prepare-code-work 절차 + +1. **task 매핑 검증** + - `task-path`의 active PLAN/review pair가 같은 task header를 갖고 아직 verdict가 없는지 확인한다. + - `agent-ui-docs`가 현재 활성 문서이고 status가 `계획`인지 확인한다. `가정`, `불명확`, 이미 archive 된 문서는 매핑하지 않는다. + - `code-paths`는 실제 구현 후보로 좁혀진 경로만 기록한다. 존재하지 않는 새 파일 후보는 경로와 생성 의도가 plan에 명시된 경우에만 허용한다. + +2. **pending entry 기록** + - `agent-ui/.sync-state.json.pending_code_work`에 아래 객체를 `task_path` 기준으로 upsert한다. + - 같은 `task_path`의 내용이 다르면 덮어쓰지 말고 현재 plan 범위와 비교해 갱신 근거를 확인한다. + +```json +{ + "task_path": "agent-task/", + "scope": "", + "agent_ui_paths": ["agent-ui/definition/.../index.md"], + "code_paths": [""], + "verification_requirements": [": "], + "prepared_at": "", + "state": "pending" +} +``` + +3. **state 경계 유지** + - `schema_version`을 `2`로 올리고 `pending_code_work` 외의 baseline 필드를 보존한다. + - 이 모드는 `last_synced_head`, agent-ui status, code evidence를 바꾸지 않고 commit/push하지 않는다. + - 일반 plan/review 문서에는 agent-ui 전용 completion section이나 review 전용 status 전환 지시를 추가하지 않는다. + +### reconcile-completion 절차 + +1. **완료 이벤트 매칭** + - exact `completion-log`가 일반 code-review PASS 또는 user-review-resolved PASS의 terminal artifact인지 확인한다. + - `task-path`가 입력되면 그 값으로, 없으면 `completion-log`의 archived task path를 원래 `agent-task/` 형태로 정규화해 `pending_code_work[].task_path`와 매칭한다. + - zero/multiple match이면 agent-ui 문서를 추측해 갱신하지 않는다. + +2. **완료 evidence 검증** + - pending entry의 모든 `code_paths`가 실제 존재하고 agent-ui 정의를 구현한 근거인지 확인한다. + - `completion-log`의 최종 검증과 `verification_requirements`가 충족됐는지 확인한다. + - generic `complete.log`에 agent-ui 전용 section이 없어도 정상으로 본다. 필요한 매핑은 pending entry에서만 읽는다. + - evidence가 부족하면 entry를 `pending`으로 유지하고 일반 plan/review artifact를 수정하지 않는다. + +3. **agent-ui 상태 반영** + - matching 문서마다 `update-agent-ui`를 `status=구현됨`, 실제 code evidence, `post-validate=false`, `post-sync=false`로 실행한다. + - 모든 문서 갱신 뒤 `validate-agent-ui`를 matching scope, `sync-intent=skip`으로 실행한다. + - 갱신이나 검증이 실패하면 pending entry를 제거하지 않고 `agent-ui/USER_REVIEW.md`에 `Source Skill: sync-agent-ui`, `Stage: completion-reconcile`로 차단 근거를 남긴다. + +4. **pending entry 종료** + - 갱신과 검증이 모두 통과한 경우에만 matching pending entry를 제거한다. + - `last_sync_mode`을 `plan-reconcile`, `last_synced_at`, `agent_ui_paths`, `code_paths`, `notes`를 완료 근거로 갱신한다. + - `sync-result-head`가 실제 commit이고 matching agent-ui/code 결과를 모두 포함하는지 검증된 경우에만 `last_synced_head`를 그 값으로 갱신한다. 그렇지 않으면 기존 `last_synced_head`를 보존하고 notes에 기준점 미갱신 사유를 남긴다. + - 이미 같은 `completion-log`가 notes에 완료 근거로 기록되어 있고 pending entry가 없으면 idempotent `ALREADY_RECONCILED`로 반환한다. + - 이 모드는 일반 plan/review/complete.log를 수정하거나 commit/push하지 않는다. + +### incremental/full 절차 + 1. **동기화 범위 확정** - `.sync-state.json`이 없고 `mode=incremental`이면 현재 agent-ui를 legacy baseline 후보로 본다. - legacy baseline 후보는 `validate-agent-ui` 통과 후 agent-ui baseline commit과 `.sync-state.json` commit을 남기는 migration으로 처리하고, 같은 실행에서 코드 반영은 하지 않는다. @@ -64,13 +131,15 @@ description: 정합화된 agent-ui 변경분을 코드에 직접 반영하거나 - `plan-required`는 목표와 범위는 명확하지만 독립 리뷰 가능한 구현 루프가 필요할 때 사용한다. - 여러 view/component에 걸친 반영, shared component 추출, route/shell 조정, 테스트/검증 보강, UI 상태와 코드 변경을 함께 다루는 작업이 여기에 속한다. - 이 경우 직접 코드 반영을 시작하지 않고 `plan` 스킬로 전환한다. - - plan에는 반영 대상 agent-ui 문서, 코드 후보, 최종 `validate-agent-ui`, code evidence 기준을 포함한다. - - plan이 만드는 `CODE_REVIEW-*-G??.md`에는 `Agent UI Completion` 섹션을 넣어 code-review PASS 때 listed agent-ui 문서를 `구현됨`으로 전환하게 한다. + - plan에는 일반 구현 범위, 코드 후보, 검증 기준만 넘긴다. + - plan pair가 생성되면 이 스킬의 `prepare-code-work`를 실행해 반영 대상 agent-ui 문서, 코드 후보, verification requirement를 task에 매핑한다. + - 일반 code-review PASS와 `complete.log` 생성 뒤 caller/router/runtime가 `reconcile-completion`을 실행해 mapped agent-ui 문서를 `구현됨`으로 전환한다. - `milestone-required`는 코드 구현이 제품 UI 구조나 장기 작업 단위까지 바꾸는 경우 사용한다. - navigation/information architecture 재정렬, 외부 제품 UI 분석 기반 재설계, 여러 workflow/role/surface를 묶는 작업, 활성 Milestone 범위 밖 작업이 여기에 속한다. - 이 경우 직접 코드 반영과 plan 생성을 시작하지 않고 `update-roadmap`으로 Milestone/Epic/Task 배치를 먼저 남긴다. - 이 Milestone에는 `완료 리뷰`의 `agent-ui 상태 반영: 대기`를 남기고, `status: 구현됨` 전환은 plan 완료가 아니라 Milestone 종료 검토 항목으로 둔다. - - `plan-required` 또는 `milestone-required`로 판정하면 agent-ui 문서 status를 `구현됨`으로 바꾸지 않고, sync state와 commit/push도 갱신하지 않는다. + - `plan-required` 판정 시 `prepare-code-work` pending entry 외에는 agent-ui 문서 status, baseline sync state, commit/push를 갱신하지 않는다. + - `milestone-required` 판정 시 agent-ui 문서 status, sync state, commit/push를 갱신하지 않는다. 3. **코드 반영** - 이 단계는 `execution-route=direct-sync`일 때만 수행한다. `plan-required` 또는 `milestone-required` 판정이면 Step 2의 라우팅 결과를 보고하고 해당 스킬 흐름으로 전환한다. @@ -111,9 +180,15 @@ description: 정합화된 agent-ui 변경분을 코드에 직접 반영하거나 - [ ] `validate-agent-ui`가 먼저 실행되었거나 `validated=true` 근거가 있는가 - [ ] 현재 sync 범위에 미해결 USER_REVIEW가 없는가 - [ ] 코드 반영 전에 `direct-sync`, `plan-required`, `milestone-required` 중 하나로 판정했는가 -- [ ] `plan-required` 또는 `milestone-required` 판정에서 직접 코드 변경, status 구현됨 전환, sync state 갱신, commit/push를 하지 않았는가 +- [ ] `plan-required` 판정에서 직접 코드 변경, status 구현됨 전환, baseline sync state 갱신, commit/push를 하지 않고 pending entry만 기록했는가 +- [ ] `milestone-required` 판정에서 직접 코드 변경, status 구현됨 전환, sync state 갱신, commit/push를 하지 않았는가 - [ ] `direct-sync` 판정에서 코드 반영과 검증이 통과한 항목은 같은 실행 안에서 agent-ui 문서 status를 `구현됨`으로 바꾸고 code evidence를 남겼는가 -- [ ] `plan-required` 판정에서 생성될 `CODE_REVIEW-*-G??.md`에 `Agent UI Completion` 섹션으로 status 전환 규칙을 넘겼는가 +- [ ] `prepare-code-work`가 exact task path와 활성 agent-ui 문서를 unique pending entry로 기록했는가 +- [ ] `prepare-code-work`가 일반 plan/review 문서에 agent-ui 전용 completion section을 요구하지 않았는가 +- [ ] `reconcile-completion`이 exact completion log, code path, verification requirement를 확인했는가 +- [ ] `reconcile-completion`이 `update-agent-ui` 다음 `validate-agent-ui sync-intent=skip` 순서로 실행했는가 +- [ ] reconcile 실패 시 pending entry를 유지하고 성공 시에만 제거했는가 +- [ ] 검증되지 않은 `sync-result-head`로 `last_synced_head`를 바꾸지 않았는가 - [ ] `가정` 또는 `불명확` 항목을 코드로 반영하지 않았는가 - [ ] `direct-sync`로 코드 반영된 항목의 status가 `구현됨`이고 code evidence가 있는가 - [ ] `direct-sync` 관련 검증이 통과했는가 @@ -127,11 +202,13 @@ description: 정합화된 agent-ui 변경분을 코드에 직접 반영하거나 ## 출력 형식 ```md -## agent-ui 코드 동기화 결과: +## agent-ui 코드 동기화 결과: -- Mode: +- Mode: - Scope: -- Execution Route: +- Execution Route: +- Task Mapping: +- Completion Log: - Agent UI Changes: <파일 목록 또는 없음> - Code Changes: <파일 목록 또는 없음> - Status Updates: <구현됨 전환 목록 또는 없음> @@ -148,6 +225,9 @@ description: 정합화된 agent-ui 변경분을 코드에 직접 반영하거나 - 미해결 USER_REVIEW가 있는 범위를 코드에 반영하지 않는다. - `가정` 또는 `불명확` 상태 항목을 코드로 구현하지 않는다. - `plan-required` 또는 `milestone-required`로 판정된 작업을 같은 `sync-agent-ui` 실행에서 직접 구현하지 않는다. +- 일반 plan/review/complete.log에 agent-ui 전용 completion schema나 status 전환 책임을 추가하지 않는다. +- matching pending entry 없이 `complete.log`만 보고 agent-ui 문서를 추측해 갱신하지 않는다. +- reconcile 실패 시 pending entry를 제거하거나 `last_synced_head`를 추정하지 않는다. - `execution-route=milestone-required`로 라우팅된 agent-ui 코드 작업의 일반 plan 완료만으로 agent-ui status를 일괄 `구현됨`으로 바꾸지 않는다. 종료 검토 통과와 code evidence가 있는 항목만 반영한다. - 검증 실패나 사용자 판단 차단 상태에서 commit/push하지 않는다. - `.sync-state.json`을 agent-ui 변경분 판정 대상으로 삼지 않는다. diff --git a/agent-ops/skills/common/update-test/SKILL.md b/agent-ops/skills/common/update-test/SKILL.md index 1585a3f..0d0e35e 100644 --- a/agent-ops/skills/common/update-test/SKILL.md +++ b/agent-ops/skills/common/update-test/SKILL.md @@ -1,7 +1,7 @@ --- name: update-test -version: 1.1.0 -description: 기존 agent-test 환경 rules.md 또는 도메인/검증 시나리오별 테스트 rule 문서를 수정하는 스킬 +version: 1.2.0 +description: 기존 agent-test 환경 rules.md 또는 테스트 profile을 수정하고, plan 같은 소비자에게 파일 수정 없는 Verification Context를 제공하는 스킬 --- # update-test @@ -10,6 +10,7 @@ description: 기존 agent-test 환경 rules.md 또는 도메인/검증 시나리 기존 `agent-test//rules.md` 또는 `agent-test//.md`를 최신 테스트 기준에 맞게 갱신한다. 라우팅은 3홉 안에 유지하고, 도메인/검증 시나리오별 문서는 자체 완결되게 보완한다. +`resolve-context` 모드에서는 해당 문서를 수정하지 않고 현재 task에 적용할 명령, 판정 기준, 제약, 외부 환경 preflight를 중립 `Verification Context`로 반환한다. `test-case`는 기존 호출과의 호환을 위한 alias이며, 새 문서 기준 이름은 `test-profile`이다. @@ -18,30 +19,42 @@ description: 기존 agent-test 환경 rules.md 또는 도메인/검증 시나리 - 기존 테스트 환경 규칙을 수정할 때 - 도메인/검증 시나리오별 테스트 기준, 명령, 판정 기준을 보완할 때 - 테스트 라우팅을 추가, 제거, 정리할 때 +- plan 또는 다른 소비자가 구현 범위에 맞는 테스트 환경/profile 정보를 필요로 할 때 ## 입력 -- `env`: `local`, `dev`, `qa`, `prod` 중 하나 (필수) +- `mode`: `update` 또는 `resolve-context`. 기본값은 `update` (선택) +- `env`: `local`, `dev`, `qa`, `prod` 중 하나. 기본값은 `local` (선택) - `test-profile`: 수정할 도메인/검증 시나리오별 테스트 문서 이름, kebab-case (선택) - `test-case`: `test-profile`의 호환 alias (선택) - `domain`: 대상 도메인 이름, kebab-case (선택) -- `change`: 수정할 내용 요약 (필수) +- `change`: 수정할 내용 요약. `mode=update`에서 필수 (선택) +- `task-summary`: 검증할 동작과 완료 조건 요약. `mode=resolve-context`에서 필수 (선택) +- `scope-paths`: 변경 후보 source/test 경로 목록. `mode=resolve-context`에서 선택 +- `verification-type`: 예: `unit`, `smoke`, `integration`, `e2e`. `mode=resolve-context`에서 선택 ## 먼저 확인할 것 - [ ] `agent-ops/rules/common/rules.md`의 스킬 규칙과 테스트 규칙이 분리되어 있는지 확인한다. - [ ] `agent-ops/skills/common/router.md`에 `update-test` 라우팅이 있는지 확인한다. - [ ] `test-case`가 있고 `test-profile`이 없으면 `test-profile`로 취급한다. -- [ ] `agent-test//rules.md` 존재 여부를 확인하고, 없으면 `create-test` 대상으로 보고 중단한다. +- [ ] `mode=update`이면 `change`가 있는지, `mode=resolve-context`이면 `task-summary`가 있는지 확인한다. +- [ ] `agent-test//rules.md` 존재 여부를 확인한다. `mode=update`에서 없으면 `create-test` 대상으로 보고 중단하고, `mode=resolve-context`에서는 `rules_state: missing`으로 반환한다. - [ ] `agent-test//rules.md`가 있으면 읽는다. -- [ ] `test-profile`이 있으면 `agent-test//.md` 존재 여부를 확인하고, 없으면 `create-test` 대상으로 보고 중단한다. +- [ ] `test-profile`이 있으면 `agent-test//.md` 존재 여부를 확인한다. `mode=update`에서 없으면 `create-test` 대상으로 보고 중단하고, `mode=resolve-context`에서는 gap으로 기록한다. - [ ] `test-profile` 문서가 있으면 읽는다. -- [ ] `test-profile`이 없고 `domain` 또는 `change`가 특정 도메인/검증 시나리오를 가리키면 env rules의 라우팅에서 대상 문서를 찾는다. +- [ ] `test-profile`이 없으면 `domain`, `change`, `task-summary`, `scope-paths`, `verification-type`을 env rules의 `## 라우팅`과 대조해 모든 matching profile을 찾는다. - [ ] 템플릿 구조 확인이 필요하면 `agent-ops/rules/common/_templates/`의 테스트 템플릿을 읽는다. - [ ] 기존 라우팅이 3홉 안에 있는지 확인한다. ## 실행 절차 +1. **모드 확정** + - `mode=resolve-context`이면 아래 `resolve-context 절차`만 수행하고 파일, `.gitignore`, `last_rule_updated_at`을 수정하지 않는다. + - `mode=update`이면 아래 `update 절차`를 수행한다. + +### update 절차 + 1. **대상 확정** - 환경 공통 규칙 변경이면 `agent-test//rules.md`만 수정한다. - 특정 도메인/검증 기준 변경이면 해당 `test-profile` 문서를 수정한다. @@ -69,6 +82,33 @@ description: 기존 agent-test 환경 rules.md 또는 도메인/검증 시나리 - 보존한 환경값 - 남은 확인 필요 항목 +### resolve-context 절차 + +1. **환경 규칙 상태 판정** + - env rules를 `missing`, `blank-skeleton`, `structured-incomplete`, `usable` 중 하나로 판정한다. + - `missing` 또는 `blank-skeleton`이어도 생성하거나 보완하지 않는다. + - 실제 파일에 없는 명령, endpoint, credential, 판정 기준을 추측하지 않는다. + +2. **matching profile 해석** + - 명시된 `test-profile` 또는 env rules의 `## 라우팅`에서 `domain / verification-type / scope`가 맞는 모든 profile을 선택한다. + - 매칭 근거가 없는 profile은 읽거나 반환하지 않는다. + - route가 가리키는 profile이 없거나 구조적으로 비어 있으면 gap으로 기록한다. + +3. **검증 사실 추출** + - env rules와 matching profile에서 실행 위치, 명령, 필수 순서, 기대 결과, 차단 기준, cache 허용 여부, 외부 서비스/secret 요구 여부를 원문 의미를 바꾸지 않고 추출한다. + - 명령이 여러 profile에 걸치면 profile별 출처와 적용 범위를 유지한다. + - `<확인 필요>` 또는 서로 충돌하는 값은 확정하지 않고 gap으로 기록한다. + +4. **외부 환경 preflight 구성** + - 현재 checkout 밖의 runner, field/bootstrap, external provider, Docker/code-server, emulator/device, shared runtime이 필요한 검증만 read-only preflight 대상으로 삼는다. + - env/profile에 적힌 범위 안에서 repo root/workdir, branch/HEAD/dirty state, source sync, binary/artifact, command help/version, config path, runtime identity, ports/process, external host, OS/arch 확인 항목을 구성한다. + - 안전한 read-only probe를 현재 환경에서 실행할 수 있으면 실제 결과를 반환하고, 실행할 수 없으면 필요한 probe와 blocker를 구분해 반환한다. + - secret, token, credential 원문은 읽거나 출력하지 않는다. + +5. **중립 handoff 반환** + - 소비 스킬 이름이나 문서 section 이름에 결합하지 않고 아래 출력 형식을 그대로 반환한다. + - test rule 유지보수는 사용자가 요청했거나 확인된 구조 결함이 있을 때만 `create-test` 또는 `update-test` 후보로 표시한다. `resolve-context` 실행 중 직접 호출하거나 수정하지 않는다. + ## 실행 결과 검증 - [ ] 수정 대상 문서가 여전히 필수 섹션을 포함하는가 @@ -76,6 +116,10 @@ description: 기존 agent-test 환경 rules.md 또는 도메인/검증 시나리 - [ ] env rules 라우팅이 3홉 제한을 넘기지 않는가 - [ ] 도메인/검증 시나리오별 문서가 다른 테스트 문서로 라우팅하지 않는가 - [ ] local 수정 시 `.gitignore`에 local 경로가 반영되었는가 +- [ ] `resolve-context`에서 파일과 `.gitignore`를 수정하지 않았는가 +- [ ] `resolve-context`가 읽은 env/profile 경로와 각 명령의 출처를 정확히 반환하는가 +- [ ] 외부 검증이 있으면 read-only preflight 결과 또는 실행 불가 blocker가 구분되어 있는가 +- [ ] 누락, `<확인 필요>`, 충돌 값을 확정된 명령이나 기준으로 반환하지 않았는가 - [ ] secret, token, 개인 endpoint 원문이 tracked 파일에 추가되지 않았는가 - 검증 실패 시: 문제 문서만 다시 보완한다. @@ -91,9 +135,35 @@ description: 기존 agent-test 환경 rules.md 또는 도메인/검증 시나리 - 확인 필요: <항목 또는 없음> ``` +`mode=resolve-context`: + +```md +## Verification Context + +- Environment: +- Rules State: +- Sources Read: + - +- Scope Match: + - +- Commands: + - ``: source=``; scope=``; expected=``; cache=`` +- Preconditions: + - <필수 순서, 실행 위치, config/runtime 요구 또는 없음> +- Read-only Preflight: + - ``: ; +- Constraints: + - <금지 사항, 외부 서비스/secret 요구 여부 또는 없음> +- Gaps: + - `, conflict 또는 없음> +- Maintenance: ; <근거> +``` + ## 금지 사항 - 기존 local 환경값을 추측으로 바꾸지 않는다. - 확인되지 않은 테스트 명령을 필수 검증으로 단정하지 않는다. - 도메인/검증 시나리오별 문서를 router처럼 쓰지 않는다. +- `resolve-context`에서 테스트 문서 생성, 수정, `.gitignore` 갱신을 수행하지 않는다. +- `resolve-context` 출력에 특정 소비 스킬 전용 필드나 파일명을 강제하지 않는다. - secret, token, 개인 endpoint 원문을 tracked 파일에 기록하지 않는다. 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 32d864a..6840d11 100644 --- a/agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md +++ b/agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md @@ -1,6 +1,6 @@ --- name: orchestrate-agent-task-loop -description: Run agent-task work and autonomously execute active PLAN/CODE_REVIEW loops on request. Use when dispatching dependency-ready work in parallel by task number and predecessor complete.log, running lane/G-specific Codex, Claude, agy, and Pi workers, adding Pi self-checks, converging official Codex reviews, and escalating cloud context until the task loop finishes. +description: Run agent-task work and autonomously execute active PLAN/CODE_REVIEW loops on request. Use when dispatching dependency-ready work in parallel by predecessor completion and workspace write claims, running lane/G-specific Codex, Claude, agy, and Pi workers, adding Pi self-checks, converging official Codex reviews, and escalating cloud context until the task loop finishes. --- # Orchestrate Agent Task Loop @@ -82,11 +82,13 @@ Concurrency limits: - Pi `ornith:35b`: 3. - agy: 1. - Official Codex review: no separate numeric limit. -- Run worker/self-check and official review in parallel when they belong to different dependency-ready tasks, regardless of write-set. Prevent only duplicate execution of the same task. +- Run worker/self-check and official review in parallel only when they belong to different dependency-ready tasks and their canonical PLAN write sets do not collide in the current physical workspace. Prevent duplicate execution of the same task. - Even with `complete.log`, treat an explicit predecessor as unfinished while live model/review execution evidence for that task remains. Delay only its consumers; do not propagate the delay to dependency-free siblings or other task groups. -- Run official reviews for different dependency-ready tasks in parallel. +- Run official reviews for different dependency-ready tasks with disjoint workspace claims in parallel. - Before the first review batch, normalize the Agent-Ops-managed `.gitignore` block once so reviews do not concurrently modify the same shared control file. -- Treat `Modified Files Summary` (and legacy `수정 파일 요약`) as review-scope and stagnation evidence, not as a dispatch-order constraint. +- Require exactly one valid, non-empty `Modified Files Summary` (and legacy `수정 파일 요약`) in the active or recovery PLAN. Fail the task closed when any path is broad, outside the workspace, a directory, malformed, or missing. +- Atomically claim every canonical modified-file path before admitting worker, self-check, or review. A collision is a runtime wait, not a predecessor dependency. Retain the task's claim through every stage, retry, dispatcher restart, and follow-up PLAN; replace or expand its own claim only when the new set does not collide, and release it only after verifying the completed archive. +- Scope write claims to the canonical physical workspace. Separate worktrees and clones use independent state and may run in parallel; task-group filtering never narrows the claim ledger inside one workspace. ## Prompt Contract @@ -107,7 +109,7 @@ After an AGY/Gemini worker exits `0`, apply the same `CODE_REVIEW_PATH` implemen For Pi recovery attempts, pass only `Read {PLAN_PATH}. Continue.` without a locator explanation. For other CLI escalation attempts, pass `Continue from {LOCATOR_PATH}. Check the saved context and current workspace. Keep artifact content in English. Final in Korean.` Preserve the collaboration prohibition and next-state-materialization sentence in official-review escalation and recovery prompts. Do not ask the model to write a separate handoff summary. -When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a terminal `session-stall` locator left by an earlier dispatcher, do not create a fresh session ID. Resume the prior locator's native session file with `pi --session` and the existing `--session-dir`, and pass `Think in English. Keep artifact content in English. Final in Korean. Continue this session and complete the current task.` After a dispatcher restart, find the failed locator and resume the same session. Count this same-session restart toward the same stage's 10-consecutive-failure limit. +When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a terminal `session-stall` locator left by an earlier dispatcher, first require the locator and native session to belong to the current physical workspace. Do not create a fresh session ID for an owned locator. Resume its native session file with `pi --session` and the existing `--session-dir`, and pass `Think in English. Keep artifact content in English. Final in Korean. Continue this session and complete the current task.` After a dispatcher restart, find the failed owned locator and resume the same session. Count this same-session restart toward the same stage's 10-consecutive-failure limit. ## Work-Log Contract @@ -142,12 +144,12 @@ When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a termin - If existing `state.json` cannot be read or validated as a JSON object, block the dispatcher. Never replace it with empty state or reset the 10-attempt budget. Repair or explicitly handle it before rerunning. - When a new user turn arrives, continue tracking the same overall request unless it explicitly cancels the previous request. - Let the execution layer display `작업시작`, `자가검증시작`, `리뷰시작`, `리뷰재시도`, `Pi복구재시도`, `세션응답복구재시도`, `세션연결재시도`, `리뷰결과`, `작업대기`, `작업차단`, `디스패치추적대기`, and `작업완료` directly from dispatcher stdout. Never duplicate them in model-authored `commentary`. Use `commentary` only when an attention event actually requires caller reasoning or a user decision. Event silence never grants `final`; only the two permissions in the absolute-priority section do. -- Determine every CLI's health/progress primarily from actual stdout/stderr in `stream.log`, plus native session events when available. Never use heartbeat mtime as progress evidence. Record dispatcher PID, agent PID, each process start token, and the per-attempt process environment marker in the locator. Another dispatcher must not start a duplicate attempt merely because the stream is quiet when the PID/start token or marker shows the same process is alive. For a locator without an agent PID, never infer stale state or duplicate recovery from elapsed time while any stream/native progress evidence exists; use only an actual terminal error or confirmed process exit as recovery evidence for every model. Run Pi with `--mode json` so `thinking_delta`, `text_delta`, and tool streams reach stdout. End an **exact** Pi toolCall-to-all-toolResult interval only when every `toolCall.id` in the preceding assistant event matches a later `toolResult.toolCallId`; never terminate the process on a time limit. If the locator lacks an agent PID during this interval, never classify it as stale or duplicate recovery based on log age; require recorded process evidence to show termination. Do not infer tool execution from `starting`, `unknown`, model reasoning, or post-toolResult state. Outside this interval, use only `stream.log` updates for Pi liveness; toolResult alone does not reset the model-response silence clock. If the stream stops for three minutes outside tool execution, store the final stream excerpt as `pi_silence_inspection` for Pi or `stream_silence_inspection` for another CLI, emit `모델응답점검`, and do not terminate the model process. Recover only from an actual terminal error or process exit. +- Determine every CLI's health/progress primarily from actual stdout/stderr in `stream.log`, plus native session events when available. Before accepting PID, marker, native-session, or stream evidence, require the locator path and recorded workspace identity to belong to the current physical workspace; accept an identity-less legacy locator only under the current store's `runs` root. Never use heartbeat mtime as progress evidence. Record workspace id, dispatcher PID, agent PID, each process start token, and a workspace-namespaced per-attempt process environment marker in the locator. Another dispatcher must not start a duplicate attempt merely because the stream is quiet when the PID/start token or marker shows the same process is alive. For a locator without an agent PID, never infer stale state or duplicate recovery from elapsed time while any stream/native progress evidence exists; use only an actual terminal error or confirmed process exit as recovery evidence for every model. Run Pi with `--mode json` so `thinking_delta`, `text_delta`, and tool streams reach stdout. End an **exact** Pi toolCall-to-all-toolResult interval only when every `toolCall.id` in the preceding assistant event matches a later `toolResult.toolCallId`; never terminate the process on a time limit. If the locator lacks an agent PID during this interval, never classify it as stale or duplicate recovery based on log age; require recorded process evidence to show termination. Do not infer tool execution from `starting`, `unknown`, model reasoning, or post-toolResult state. Outside this interval, use only `stream.log` updates for Pi liveness; toolResult alone does not reset the model-response silence clock. If the stream stops for three minutes outside tool execution, store the final stream excerpt as `pi_silence_inspection` for Pi or `stream_silence_inspection` for another CLI, emit `모델응답점검`, and do not terminate the model process. Recover only from an actual terminal error or process exit. - Detect a local-model `repetition-loop` only when the same normalized chunk repeats three consecutive times with no new tool event or file/state change. Do not infer it from similarity or semantic duplication in `thinking_delta`/`text_delta`. This signal alone must not terminate the process, block the task, trigger recovery/retry, or escalate the model; keep observing for substantive progress or an actual terminal error. - Keep `provider-connection`, `provider-stream-disconnect`, `session-stall`, `generic-error`, `process-terminated`, context/quota/model errors, and review-control violations distinct, but make them share a budget of 10 consecutive automatic recovery failures for the same task stage. On the 10th failure, block that task and do not auto-resume after cooldown. Reset the stage counter after success. - Record an explicit terminal blocker when Pi self-check leaves implementation fields incomplete 10 consecutive times or official review makes no change 10 consecutive times. -- While one task recovers or becomes blocked, continue every ready/running task that does not require it as a predecessor. Internal recovery or blocking must not trigger an arbitrary complete-candidate rescan. -- If review shared-state preflight fails, block only ready review tasks and still start every worker/self-check in the same pass. The complete scan after `complete.log` must preserve the existing snapshot rather than reread already running task directories, avoiding races with parallel archive moves that could stop another process. +- While one task recovers or becomes blocked, continue every ready/running task that neither requires it as a predecessor nor collides with its retained workspace claim. Internal recovery or blocking must not trigger an arbitrary complete-candidate rescan. +- If review shared-state preflight fails, block only ready review tasks and still start every worker/self-check with a disjoint claim in the same pass. The complete scan after `complete.log` must preserve the existing snapshot rather than reread already running task directories, avoiding races with parallel archive moves that could stop another process. - For KST-night `local-G07`–`local-G08` Laguna locator `context-limit`/`session-stall`, prefer the Prompt Contract's same-session resume and display `Pi세션연속재시작`. Use a fresh session and `세션응답복구재시도` only for other legacy Pi `session-stall` recovery. - Do not stop for user review based on filename alone. Recognize a `user-review` terminal blocker only when the active task's `USER_REVIEW.md` contains `상태: USER_REVIEW`, `유형: milestone-lock`, a real `agent-roadmap/**/milestones/*.md` target, non-`없음`/`미정` blocker rationale, unresolved decisions, and resume conditions that prevent the next safe implementation step. If the form is incomplete or conflicts with active PLAN/CODE_REVIEW, block it as a task-state contract error instead. - Recognize `## Code Review Result` (with `Overall Verdict: PASS|WARN|FAIL`) or legacy `## 코드리뷰 결과` (with `종합 판정: PASS|WARN|FAIL`) as the review verdict. If both canonical and legacy headings are present in the same file, fail closed. Never parse the same string in implementation evidence, command output, or example text as the runtime verdict. @@ -191,12 +193,12 @@ When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a termin ``` - If a worker/self-check/review future ends without `complete.log`, reread only that task and run its next stage. Do not rescan the complete candidate set. - - Persist `active_stage` for a running task. After dispatcher restart, exclude only that task from candidates and immediately dispatch every other dependency-ready task in parallel. A running task must not delay candidate discovery or another task's execution. - - **ABSOLUTE RULE:** Scan the complete candidate set only at initial entry and immediately after creating a verified `complete.log`. In that scan, exclude only tasks shown as running by persistent state and native session/locator evidence; immediately run all dependency-ready tasks in parallel. An unmet dependency excludes only that task. Exit instead of polling when no candidate remains. + - Persist `active_stage` for a running task. After dispatcher restart, exclude that task from candidates, restore or conservatively adopt its workspace write claim, and immediately dispatch every other dependency-ready task whose claim does not collide. + - **ABSOLUTE RULE:** Scan the complete candidate set only at initial entry and immediately after creating a verified `complete.log`. In that scan, exclude tasks shown as running by current-workspace state and native session/locator evidence, then atomically admit every dependency-ready task with a non-colliding write claim. An unmet dependency or write collision excludes only that task. Exit instead of polling when no candidate remains. - Persist Pi worker success, Pi self-check success, and official review as separate stages. If restart state is `worker_done=true` and `selfcheck_done=false`, resume with a fresh self-check session on the same Pi model, not with worker or review. - Key persistent state to the `task/plan/tag` generation at the start of PLAN. Checklist/body edits to the same PLAN do not reset the stage; a new plan number in a follow-up PLAN does. - Send an already completed review stub with no dispatcher execution record to review. Never send dispatcher-recorded Pi worker success to review before self-check completes. - - Start official review and worker/self-check together when they belong to different dependency-ready tasks. Do not wait for another task's checkout or write-set. + - Start official review and worker/self-check together when they belong to different dependency-ready tasks with disjoint workspace claims. Wait for a claim owner to reach verified completion before admitting a colliding task. - Let the dispatcher record every worker/self-check/review attempt start and finish in the task-group `WORK_LOG.md`. - Archive `WORK_LOG.md` as `work_log_N.log` only after the final task review process exits, the dispatcher appends `FINISH`, and a complete scan finds no active/running task in that group. Accept the log at either the active group path or the verified completed single-task archive; do not impose either location contract on common plan/code-review. @@ -208,12 +210,12 @@ When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a termin - Recover timeout, crash, process termination, permission, and ordinary implementation errors on the same target within the same stage's 10-consecutive-failure limit, preserving the actual failure class and locator. At exhaustion, block only that task and keep dispatching independent work. - On success after escalation, record `worker_cli` and `worker_model` from the successful locator's actual target, not the initial PLAN route. - Never escalate Pi to a cloud model. - - Use attempt identity `__p____aNN`. Record workspace, CLI/model/reasoning effort, PLAN/review, `WORK_LOG.md`, session ID, native session path, and raw output log in the locator. + - Use attempt identity `__p____aNN` and namespace the process marker with the physical workspace id. Record canonical workspace root/id, CLI/model/reasoning effort, PLAN/review, `WORK_LOG.md`, session ID, native session path, and raw output log in the locator. - Store locators under repository `.git/agent-task-dispatcher/runs/`. Fall back to `${XDG_STATE_HOME}/agent-task-dispatcher//runs/` only when `.git` state is unwritable. 4. **Converge review.** - - Run every official review in an independent Codex one-shot session with no separate numeric limit. Dispatch all ready reviews in parallel. - - For finalization recovery without an active PLAN, recover the review target from the archived plan log for the same task/plan/tag. Never treat the write-set as a dispatch blocker. + - Run every official review in an independent Codex one-shot session with no separate numeric limit. Dispatch all ready reviews with disjoint workspace claims in parallel. + - For finalization recovery without an active PLAN, recover the review target and write claim from the archived plan log for the same task/plan/tag. Keep the claim until the completed archive is verified. - Forbid collaboration/sub-agent tools in official review and finish inside the current one-shot session. If such a tool call appears, clean up that attempt's independent subprocess group and retry in a fresh review session. Count the failure toward the same stage's 10-consecutive-failure limit. - Delegate PASS archive, WARN/FAIL follow-up pairs, and review-finalization recovery to the `code-review` file contract. - Reclassify any remaining active pair and send it to worker or review. @@ -222,10 +224,10 @@ When recovering a KST-night `local-G07`–`local-G08` Laguna locator or a termin ## Verification Checklist -- [ ] Scan the complete candidate set only on initial entry and immediately after verified `complete.log`; start every ready candidate except running tasks in parallel in the same pass. +- [ ] Scan the complete candidate set only on initial entry and immediately after verified `complete.log`; atomically claim and start every non-running, dependency-ready, non-colliding candidate in the same pass. - [ ] Confirm the actual CLI/model for each route matches the routing table. - [ ] Run exactly one fresh-session self-check only for Pi work. -- [ ] Run every official review with Codex `gpt-5.6-sol` xhigh and dispatch dependency-ready reviews in parallel without a numeric limit. +- [ ] Run every official review with Codex `gpt-5.6-sol` xhigh and dispatch dependency-ready reviews with disjoint workspace claims in parallel without a numeric limit. - [ ] Locate the native session and output log for every attempt locator. - [ ] Record every worker/self-check/review attempt `START`/`FINISH` in one task-group `WORK_LOG.md`. - [ ] For every completed task group that generated `WORK_LOG.md`, archive a `work_log_N.log` containing the final review `FINISH` and leave no active `WORK_LOG.md`. 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 fb190a0..cb8d00a 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 @@ -438,8 +438,11 @@ def next_execution_identity( class StateStore: def __init__(self, workspace: Path): - workspace_id = hashlib.sha256(str(workspace).encode()).hexdigest()[:16] - git_marker = workspace / ".git" + self.workspace = workspace.resolve() + self.workspace_id = hashlib.sha256( + str(self.workspace).encode() + ).hexdigest()[:16] + git_marker = self.workspace / ".git" git_directory: Path | None = None if git_marker.is_dir(): git_directory = git_marker @@ -447,12 +450,16 @@ class StateStore: marker = git_marker.read_text(encoding="utf-8", errors="replace").strip() if marker.startswith("gitdir:"): candidate = Path(marker.split(":", 1)[1].strip()) - git_directory = candidate if candidate.is_absolute() else (workspace / candidate).resolve() + git_directory = ( + candidate + if candidate.is_absolute() + else (self.workspace / candidate).resolve() + ) candidates = [] if git_directory is not None: candidates.append(git_directory / "agent-task-dispatcher") state_base = Path(os.environ.get("XDG_STATE_HOME", str(Path.home() / ".local" / "state"))) - candidates.append(state_base / "agent-task-dispatcher" / workspace_id) + candidates.append(state_base / "agent-task-dispatcher" / self.workspace_id) self.root = candidates[-1] last_error: OSError | None = None for candidate in candidates: @@ -508,6 +515,128 @@ class StateStore: ) else: self.data = {"tasks": {}, "attempt_counters": {}} + try: + self._bind_workspace_identity() + self.write_claim_snapshot() + except DispatcherTerminalStateError: + self.lock_stream.close() + raise + + def _bind_workspace_identity(self) -> None: + expected = { + "id": self.workspace_id, + "root": str(self.workspace), + } + current = self.data.get("workspace_identity") + if current is None: + self.data["workspace_identity"] = expected + return + if not isinstance(current, dict): + raise DispatcherTerminalStateError( + f"dispatcher workspace identity가 object가 아니다: {self.path}" + ) + if ( + current.get("id") != expected["id"] + or current.get("root") != expected["root"] + ): + raise DispatcherTerminalStateError( + "dispatcher state의 workspace identity가 현재 checkout과 다르다: " + f"state={current} current={expected}" + ) + + def write_claim_snapshot(self) -> dict[str, dict[str, Any]]: + raw = self.data.setdefault("write_claims", {}) + if not isinstance(raw, dict): + raise DispatcherTerminalStateError( + f"dispatcher write_claims가 object가 아니다: {self.path}" + ) + snapshot: dict[str, dict[str, Any]] = {} + for owner, value in raw.items(): + if not isinstance(owner, str) or not owner or not isinstance(value, dict): + raise DispatcherTerminalStateError( + f"dispatcher write claim 형식이 유효하지 않다: owner={owner!r}" + ) + paths = value.get("paths") + exclusive = value.get("exclusive", False) + if not isinstance(paths, list) or not isinstance(exclusive, bool): + raise DispatcherTerminalStateError( + f"dispatcher write claim 경로 형식이 유효하지 않다: owner={owner}" + ) + if value.get("workspace_id") != self.workspace_id: + raise DispatcherTerminalStateError( + "dispatcher write claim의 workspace identity가 다르다: " + f"owner={owner}" + ) + canonical: list[str] = [] + for raw_path in paths: + if not isinstance(raw_path, str) or not raw_path: + raise DispatcherTerminalStateError( + f"dispatcher write claim 경로가 유효하지 않다: owner={owner}" + ) + path = Path(raw_path) + resolved = path.resolve() + try: + resolved.relative_to(self.workspace) + except ValueError as exc: + raise DispatcherTerminalStateError( + "dispatcher write claim이 workspace 밖을 가리킨다: " + f"owner={owner} path={raw_path}" + ) from exc + if not path.is_absolute() or str(resolved) != raw_path or resolved == self.workspace: + raise DispatcherTerminalStateError( + "dispatcher write claim 경로가 canonical file이 아니다: " + f"owner={owner} path={raw_path}" + ) + canonical.append(raw_path) + if (not canonical and not exclusive) or len(canonical) != len(set(canonical)): + raise DispatcherTerminalStateError( + f"dispatcher write claim 경로 집합이 유효하지 않다: owner={owner}" + ) + record = dict(value) + record["paths"] = sorted(canonical) + snapshot[owner] = record + return snapshot + + def replace_write_claims( + self, + claims: dict[str, dict[str, Any]], + *, + persist: bool, + ) -> None: + previous = self.data.get("write_claims", {}) + self.data["write_claims"] = claims + try: + self.write_claim_snapshot() + if persist and previous != claims: + self.save() + except BaseException: + self.data["write_claims"] = previous + raise + + def adopt_active_write_claim(self, task: Task) -> None: + claims = self.write_claim_snapshot() + if task.name in claims: + return + timestamp = now_iso() + claims[task.name] = { + "task": task.name, + "plan_hash": task.plan_hash, + "paths": sorted(task.write_set) if task.write_set_known else [], + "exclusive": not task.write_set_known, + "workspace_id": self.workspace_id, + "acquired_at": timestamp, + "updated_at": timestamp, + "source": "active-recovery", + } + self.replace_write_claims(claims, persist=True) + + def release_write_claim(self, task_name: str, *, persist: bool = True) -> bool: + claims = self.write_claim_snapshot() + if task_name not in claims: + return False + del claims[task_name] + self.replace_write_claims(claims, persist=persist) + return True def save(self) -> None: write_json(self.path, self.data) @@ -829,6 +958,7 @@ class StateStore: ) record.update(status="complete", archive=str(archive_path)) record.pop("reason", None) + self.release_write_claim(task_name, persist=False) self.save() cleanup_completed_task_attempt_logs(self.runs, task_name) @@ -873,6 +1003,11 @@ class StateStore: archive = str(record.get("archive") or "") if archive and (Path(archive) / "complete.log").is_file(): completed[task_name] = archive + if task_name not in active_or_running: + changed = ( + self.release_write_claim(task_name, persist=False) + or changed + ) else: errors[task_name] = "persisted complete archive가 유효하지 않다" continue @@ -888,6 +1023,7 @@ class StateStore: archive = str(candidates[0].resolve()) record.update(status="complete", archive=archive) completed[task_name] = archive + changed = self.release_write_claim(task_name, persist=False) or changed changed = True elif not candidates: errors[task_name] = ( @@ -928,7 +1064,12 @@ def orchestration_live_agent_processes( state = task_states.get(task_name) if not isinstance(state, dict): continue - is_live, detail = external_active_is_live(state) + is_live, detail = external_active_is_live( + state, + expected_workspace=store.workspace, + expected_workspace_id=store.workspace_id, + expected_runs_root=store.runs, + ) if is_live: live[task_name] = detail return live @@ -950,6 +1091,7 @@ def parse_task_name(task_root: Path, directory: Path) -> str: def extract_write_set(plan: Path | None, workspace: Path) -> tuple[set[str], bool]: if plan is None or not plan.exists(): return set(), False + workspace = workspace.resolve() text = plan.read_text(encoding="utf-8", errors="replace") matches = [] for heading in MODIFIED_FILES_HEADINGS: @@ -970,7 +1112,10 @@ def extract_write_set(plan: Path | None, workspace: Path) -> tuple[set[str], boo for value in re.findall(r"`([^`]+)`", cells[0]): normalized = re.sub(r":\d+(?::\d+)?$", "", value.strip()) if normalized and not normalized.startswith(("http://", "https://")): - if any(character in normalized for character in "*?[]"): + if ( + any(character in normalized for character in "*?[]") + or normalized.endswith(("/", "\\")) + ): invalid = True continue candidate = Path(normalized) @@ -1062,6 +1207,12 @@ def read_task_directory(workspace: Path, directory: Path) -> Task | None: recovery_plan = matching_plan_log(directory, recovery_log) write_set_source = plan or recovery_plan write_set, write_set_known = extract_write_set(write_set_source, workspace) + if (write_set_source is not None and not write_set_known) or ( + recovery_log is not None and recovery_plan is None + ): + errors.append( + "PLAN Modified Files Summary가 없거나 비어 있거나 유효하지 않다" + ) if plan is not None: metadata = PLAN_IDENTITY_RE.search( plan.read_text(encoding="utf-8", errors="replace")[:1024] @@ -2922,18 +3073,117 @@ def marked_agent_process_pids(marker: str) -> list[int]: return sorted(matches) -def external_active_is_live(state: dict[str, Any]) -> tuple[bool, str]: +def locator_workspace_ownership( + locator_path: Path, + locator: dict[str, Any], + *, + expected_workspace: Path | None = None, + expected_workspace_id: str | None = None, + expected_runs_root: Path | None = None, +) -> tuple[bool, str]: + if expected_workspace is None and expected_workspace_id is None: + return True, "" + try: + expected_root = ( + expected_workspace.resolve() if expected_workspace is not None else None + ) + expected_id = expected_workspace_id + if expected_id is None and expected_root is not None: + expected_id = hashlib.sha256(str(expected_root).encode()).hexdigest()[:16] + if expected_runs_root is None: + return False, "현재 workspace의 locator runs root가 없다" + resolved_runs = expected_runs_root.resolve() + resolved_locator = locator_path.resolve() + resolved_locator.relative_to(resolved_runs) + except (OSError, RuntimeError, ValueError): + return ( + False, + "foreign workspace locator path: " + f"locator={locator_path} expected_runs={expected_runs_root}", + ) + + recorded_workspace = locator.get("workspace") + recorded_workspace_id = locator.get("workspace_id") + if recorded_workspace_id not in (None, "") and ( + str(recorded_workspace_id) != str(expected_id) + ): + return ( + False, + "foreign workspace locator id: " + f"recorded={recorded_workspace_id} expected={expected_id}", + ) + if recorded_workspace not in (None, ""): + try: + recorded_root = Path(str(recorded_workspace)).resolve() + except (OSError, RuntimeError): + return False, "locator workspace 경로를 canonicalize할 수 없다" + if expected_root is not None and recorded_root != expected_root: + return ( + False, + "foreign workspace locator root: " + f"recorded={recorded_root} expected={expected_root}", + ) + evidence_fields = ["stream_log"] + if locator.get("cli") == "pi": + evidence_fields.append("native_session_path") + for field in evidence_fields: + raw_evidence = locator.get(field) + if raw_evidence in (None, ""): + continue + try: + Path(str(raw_evidence)).resolve().relative_to(resolved_runs) + except (OSError, RuntimeError, ValueError): + return ( + False, + "foreign workspace locator evidence: " + f"field={field} path={raw_evidence}", + ) + # An identity-less legacy locator is accepted only because physical + # containment under the current store's runs root was already proved. + return True, "" + + +def external_active_is_live( + state: dict[str, Any], + *, + expected_workspace: Path | None = None, + expected_workspace_id: str | None = None, + expected_runs_root: Path | None = None, +) -> tuple[bool, str]: raw_locator = state.get("active_locator") if not raw_locator: return False, "active locator 없음" target = Path(str(raw_locator)) locator_path = target if target.name == "locator.json" else target / "locator.json" + path_owned, ownership_detail = locator_workspace_ownership( + locator_path, + {}, + expected_workspace=expected_workspace, + expected_workspace_id=expected_workspace_id, + expected_runs_root=expected_runs_root, + ) + if not path_owned: + return False, ownership_detail locator: dict[str, Any] = {} if locator_path.is_file(): try: locator = json.loads(locator_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return False, f"locator 판독 실패: {locator_path}" + if not isinstance(locator, dict): + return False, f"locator object 형식이 아니다: {locator_path}" + + owned, ownership_detail = locator_workspace_ownership( + locator_path, + locator, + expected_workspace=expected_workspace, + expected_workspace_id=expected_workspace_id, + expected_runs_root=expected_runs_root, + ) + if not owned: + return False, ownership_detail + + if locator: status = str(locator.get("status") or "") if status and status != "running": return False, f"locator status={status}" @@ -3026,15 +3276,42 @@ def external_active_is_live(state: dict[str, Any]) -> tuple[bool, str]: return False, f"active 증거 없음: {raw_locator}" -def laguna_resume_locator(state: dict[str, Any]) -> Path | None: +def laguna_resume_locator( + state: dict[str, Any], + *, + expected_workspace: Path | None = None, + expected_workspace_id: str | None = None, + expected_runs_root: Path | None = None, +) -> Path | None: raw_locator = state.get("active_locator") if not raw_locator: return None - locator = Path(str(raw_locator)) + target = Path(str(raw_locator)) + locator = target if target.name == "locator.json" else target / "locator.json" + path_owned, _ = locator_workspace_ownership( + locator, + {}, + expected_workspace=expected_workspace, + expected_workspace_id=expected_workspace_id, + expected_runs_root=expected_runs_root, + ) + if not path_owned: + return None try: record = json.loads(locator.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None + if not isinstance(record, dict): + return None + owned, _ = locator_workspace_ownership( + locator, + record, + expected_workspace=expected_workspace, + expected_workspace_id=expected_workspace_id, + expected_runs_root=expected_runs_root, + ) + if not owned: + return None if ( record.get("cli") != "pi" or not str(record.get("model", "")).startswith("laguna-s") @@ -3046,6 +3323,11 @@ def laguna_resume_locator(state: dict[str, Any]) -> Path | None: native = Path(str(native_raw)) if native_raw else None if native is None or not native.exists(): return None + if expected_runs_root is not None: + try: + native.resolve().relative_to(expected_runs_root.resolve()) + except (OSError, RuntimeError, ValueError): + return None return locator @@ -3128,21 +3410,54 @@ async def invoke( normalized_output_path.touch() heartbeat_path.touch() session_id = str(uuid.uuid4()) - process_marker = f"{identity}__{uuid.uuid4()}" + process_marker = f"w{store.workspace_id}__{identity}__{uuid.uuid4()}" pi_resume_session: Path | None = None if spec.local_pi and resume_locator and resume_locator.is_file(): - try: - prior = json.loads(resume_locator.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): + resume_locator_path = ( + resume_locator + if resume_locator.name == "locator.json" + else resume_locator / "locator.json" + ) + path_owned, _ = locator_workspace_ownership( + resume_locator_path, + {}, + expected_workspace=store.workspace, + expected_workspace_id=store.workspace_id, + expected_runs_root=store.runs, + ) + if path_owned: + try: + prior = json.loads(resume_locator_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + prior = {} + else: prior = {} - prior_native = prior.get("native_session_path") - candidate = Path(str(prior_native)) if prior_native else None - if candidate and candidate.is_dir(): - sessions = list(candidate.glob("*.jsonl")) - candidate = max(sessions, key=lambda path: path.stat().st_mtime_ns) if sessions else None - if candidate and candidate.is_file(): - pi_resume_session = candidate - session_id = str(prior.get("session_id") or candidate.stem) + owned, _ = locator_workspace_ownership( + resume_locator_path, + prior if isinstance(prior, dict) else {}, + expected_workspace=store.workspace, + expected_workspace_id=store.workspace_id, + expected_runs_root=store.runs, + ) + if owned and isinstance(prior, dict): + prior_native = prior.get("native_session_path") + candidate = Path(str(prior_native)) if prior_native else None + if candidate and candidate.is_dir(): + sessions = list(candidate.glob("*.jsonl")) + candidate = ( + max(sessions, key=lambda path: path.stat().st_mtime_ns) + if sessions + else None + ) + if candidate and candidate.is_file(): + try: + candidate.resolve().relative_to(store.runs.resolve()) + except (OSError, RuntimeError, ValueError): + candidate = None + if candidate and candidate.is_file(): + pi_resume_session = candidate + resume_locator = resume_locator_path + session_id = str(prior.get("session_id") or candidate.stem) started_at = now_iso() work_log_path = milestone_work_log_path(task) record: dict[str, Any] = { @@ -3151,7 +3466,8 @@ async def invoke( "plan_number": plan_number(task), "role": role, "attempt": attempt, - "workspace": str(workspace), + "workspace": str(store.workspace), + "workspace_id": store.workspace_id, **dispatcher_source_provenance(), "cli": spec.cli, "model": spec.model, @@ -5157,14 +5473,99 @@ def status_lines( def select_dispatch_candidates( + store: StateStore, ready: list[tuple[Task, str]], -) -> tuple[list[tuple[Task, str]], list[tuple[Task, str]], str]: + *, + persist: bool, +) -> tuple[ + list[tuple[Task, str]], + list[tuple[Task, str, str]], + str, +]: ready_reviews = [(task, stage) for task, stage in ready if stage == "review"] ready_workers = [(task, stage) for task, stage in ready if stage in {"worker", "selfcheck"}] - # Each scheduler pass admits every dependency-ready task. The running map - # prevents only duplicate attempts for the same task; phases and write-sets - # never impose a cross-task barrier. - return ready_reviews + ready_workers, [], "" + ordered = ready_reviews + ready_workers + claims = store.write_claim_snapshot() + selected: list[tuple[Task, str]] = [] + deferred: list[tuple[Task, str, str]] = [] + timestamp = now_iso() + for task, stage in ordered: + if not task.write_set_known or not task.write_set: + deferred.append( + ( + task, + stage, + "valid non-empty Modified Files Summary write claim이 필요하다", + ) + ) + continue + requested = sorted(task.write_set) + invalid_path: str | None = None + for raw_path in requested: + path = Path(raw_path) + resolved = path.resolve() + try: + resolved.relative_to(store.workspace) + except ValueError: + invalid_path = raw_path + break + if ( + not path.is_absolute() + or str(resolved) != raw_path + or resolved == store.workspace + ): + invalid_path = raw_path + break + if invalid_path is not None: + deferred.append( + ( + task, + stage, + f"write claim 경로가 canonical workspace file이 아니다: {invalid_path}", + ) + ) + continue + + conflict: tuple[str, str] | None = None + requested_set = set(requested) + for owner in sorted(claims): + if owner == task.name: + continue + other = claims[owner] + if other.get("exclusive"): + conflict = (owner, "") + break + intersection = sorted(requested_set & set(other.get("paths", []))) + if intersection: + conflict = (owner, intersection[0]) + break + if conflict is not None: + owner, path = conflict + deferred.append( + ( + task, + stage, + f"write claim 충돌 대기: owner={owner}; path={path}", + ) + ) + continue + + previous = claims.get(task.name, {}) + claims[task.name] = { + "task": task.name, + "plan_hash": task.plan_hash, + "paths": requested, + "exclusive": False, + "workspace_id": store.workspace_id, + "acquired_at": previous.get("acquired_at") or timestamp, + "updated_at": timestamp, + "source": "plan", + } + selected.append((task, stage)) + + if persist: + store.replace_write_claims(claims, persist=True) + return selected, deferred, "" def ensure_review_shared_state(workspace: Path) -> None: @@ -5606,7 +6007,12 @@ async def dispatch_with_store( stage = task_stage(task, state) if state.get("active_stage"): active_stage = str(state["active_stage"]) - active_live, active_detail = external_active_is_live(state) + active_live, active_detail = external_active_is_live( + state, + expected_workspace=store.workspace, + expected_workspace_id=store.workspace_id, + expected_runs_root=store.runs, + ) if active_live: reason = f"외부 실행중: stage={active_stage}; {active_detail}" active_key = ( @@ -5614,12 +6020,19 @@ async def dispatch_with_store( ) externally_active.append((task, active_stage)) blocked_details[task.name] = ("작업중", stage, reason) + if not args.dry_run: + store.adopt_active_write_claim(task) if not args.dry_run and last_wait.get(task.name) != active_key: banner("작업중", task.name, status_lines(task, stage, reason)) last_wait[task.name] = active_key continue if not args.dry_run: - resume_locator = laguna_resume_locator(state) + resume_locator = laguna_resume_locator( + state, + expected_workspace=store.workspace, + expected_workspace_id=store.workspace_id, + expected_runs_root=store.runs, + ) if resume_locator is not None: resume_locators[task.name] = resume_locator banner( @@ -5653,7 +6066,11 @@ async def dispatch_with_store( wait_key = f"{stage}|{reason}" event = ( "작업차단" - if stage in {"blocked", "user-review"} or state.get("blocked") + if ( + task.errors + or stage in {"blocked", "user-review"} + or state.get("blocked") + ) else "작업대기" ) blocked_details[task.name] = (event, stage, reason) @@ -5669,8 +6086,30 @@ async def dispatch_with_store( admission_time = datetime.now(KST) if args.dry_run: - ready_by_name = {task.name: stage for task, stage in ready} - batch_snapshot = build_admission_batch_snapshot(store, ready, admission_time) + candidates, deferred, _ = select_dispatch_candidates( + store, + ready, + persist=False, + ) + for task, stage, reason in deferred: + event = ( + "작업차단" + if reason.startswith( + ( + "valid non-empty", + "write claim 경로가", + ) + ) + else "작업대기" + ) + blocked_details[task.name] = (event, stage, reason) + waiting_tasks.append(task.name) + ready_by_name = {task.name: stage for task, stage in candidates} + batch_snapshot = build_admission_batch_snapshot( + store, + candidates, + admission_time, + ) for task in tasks: if task.name in ready_by_name: stage = ready_by_name[task.name] @@ -5722,14 +6161,30 @@ async def dispatch_with_store( else: event, stage, reason = blocked_details[task.name] banner(event, task.name, status_lines(task, stage, reason)) - return 2 if waiting_tasks and not ready else 0 + return 2 if waiting_tasks and not candidates else 0 - candidates, deferred, phase_wait_reason = select_dispatch_candidates(ready) + candidates, deferred, _ = select_dispatch_candidates( + store, + ready, + persist=True, + ) batch_snapshot = build_admission_batch_snapshot(store, candidates, admission_time) - for task, stage in deferred: - wait_key = f"{stage}|{phase_wait_reason}" + for task, stage, reason in deferred: + event = ( + "작업차단" + if reason.startswith( + ( + "valid non-empty", + "write claim 경로가", + ) + ) + else "작업대기" + ) + blocked_details[task.name] = (event, stage, reason) + waiting_tasks.append(task.name) + wait_key = f"{stage}|{reason}" if last_wait.get(task.name) != wait_key: - banner("작업대기", task.name, status_lines(task, stage, phase_wait_reason)) + banner(event, task.name, status_lines(task, stage, reason)) last_wait[task.name] = wait_key if ( 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 e1565a8..efb9765 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 @@ -123,7 +123,14 @@ class TaskStageTest(unittest.TestCase): def make_task(self, root: Path, review_text: str = ""): plan = root / "PLAN-local-G05.md" review = root / "CODE_REVIEW-local-G05.md" - plan.write_text("\n", encoding="utf-8") + target = (root / "src" / "test.py").resolve() + plan.write_text( + "\n" + "## Modified Files Summary\n\n" + "| File | Item |\n|---|---|\n" + f"| `{target}` | TEST-1 |\n", + encoding="utf-8", + ) review.write_text("\n" + review_text, encoding="utf-8") return dispatch.Task( name="test", @@ -132,6 +139,8 @@ class TaskStageTest(unittest.TestCase): review=review, user_review=None, recovery=False, + write_set={str(target)}, + write_set_known=True, lane="local", grade=5, ) @@ -441,7 +450,13 @@ class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase): directory = workspace / "agent-task" / "completing_target_test" directory.mkdir(parents=True, exist_ok=True) header = f"\n" - (directory / f"PLAN-{lane}-G{grade:02d}.md").write_text(header, encoding="utf-8") + (directory / f"PLAN-{lane}-G{grade:02d}.md").write_text( + header + + "## Modified Files Summary\n\n" + "| File | Item |\n|---|---|\n" + "| `src/completing-target.py` | TEST-1 |\n", + encoding="utf-8", + ) (directory / f"CODE_REVIEW-{lane}-G{grade:02d}.md").write_text(header, encoding="utf-8") tasks = dispatch.scan_tasks(workspace, None) return tasks[0] @@ -1831,9 +1846,11 @@ class WorkLogInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase): ) self.assertTrue( record["agent_process_marker"].startswith( - "test__p0__worker__a00__" + f"w{store.workspace_id}__test__p0__worker__a00__" ) ) + self.assertEqual(record["workspace"], str(workspace.resolve())) + self.assertEqual(record["workspace_id"], store.workspace_id) self.assertEqual(record["status"], "succeeded") prompt = build_command.call_args.args[1] self.assertEqual(prompt, "Read the plan.") @@ -2334,12 +2351,16 @@ class WorkLogInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase): (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) - native = workspace / "prior-session.jsonl" + prior_attempt = store.runs / "prior-attempt" + prior_attempt.mkdir() + native = prior_attempt / "prior-session.jsonl" native.write_text(pi_session_jsonl([]), encoding="utf-8") - prior_locator = workspace / "prior-locator.json" + prior_locator = prior_attempt / "locator.json" prior_locator.write_text( json.dumps( { + "workspace": str(workspace.resolve()), + "workspace_id": store.workspace_id, "session_id": "resume-session", "native_session_path": str(native), } @@ -2394,6 +2415,79 @@ class WorkLogInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase): record["resumed_from_locator"], str(prior_locator) ) + async def test_invoke_starts_fresh_session_for_foreign_pi_resume_locator(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) / "current" + workspace.mkdir() + (workspace / ".git").mkdir() + task = TaskStageTest().make_task(workspace) + store = dispatch.StateStore(workspace) + foreign_attempt = Path(temporary) / "foreign-attempt" + foreign_attempt.mkdir() + foreign_native = foreign_attempt / "session.jsonl" + foreign_native.write_text(pi_session_jsonl([]), encoding="utf-8") + foreign_locator = foreign_attempt / "locator.json" + foreign_locator.write_text( + json.dumps( + { + "workspace": str((Path(temporary) / "foreign").resolve()), + "workspace_id": "foreign-workspace", + "session_id": "foreign-session", + "native_session_path": str(foreign_native), + } + ), + encoding="utf-8", + ) + + def command_for( + spec, + prompt, + cwd, + actual_session_id, + attempt_dir, + pi_resume_session=None, + ): + self.assertIsNone(pi_resume_session) + self.assertNotEqual(actual_session_id, "foreign-session") + return [ + sys.executable, + "-c", + "print('fresh session', flush=True)", + ] + + spec = dispatch.AgentSpec( + "pi", + "laguna-s:2.1", + "pi", + local_pi=True, + ) + try: + with mock.patch.object( + dispatch, + "build_command", + side_effect=command_for, + ): + rc, failure, locator = await dispatch.invoke( + workspace, + store, + task, + "review", + spec, + "Continue.", + resume_locator=foreign_locator, + ) + finally: + store.close() + + self.assertEqual(rc, 0) + self.assertIsNone(failure) + record = json.loads(locator.read_text(encoding="utf-8")) + self.assertIsNone(record["resumed_from_locator"]) + self.assertNotEqual( + record["native_session_path"], + str(foreign_native), + ) + async def test_provider_stderr_requires_and_preserves_exact_evidence(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) @@ -4239,8 +4333,12 @@ class BlockerDrainTest(unittest.IsolatedAsyncioTestCase): def runnable_task(name, directory, index, deps=()): plan = directory / "PLAN-local-G05.md" review = directory / "CODE_REVIEW-local-G05.md" + target = (workspace / "src" / f"task-{index}.py").resolve() plan.write_text( - f"\n", + f"\n" + "## Modified Files Summary\n\n" + "| File | Item |\n|---|---|\n" + f"| `{target}` | TEST-1 |\n", encoding="utf-8", ) review.write_text("", encoding="utf-8") @@ -4253,6 +4351,8 @@ class BlockerDrainTest(unittest.IsolatedAsyncioTestCase): recovery=False, index=index, deps=deps, + write_set={str(target)}, + write_set_known=True, lane="local", grade=5, plan_hash=f"{name}-hash", @@ -4621,34 +4721,158 @@ class BlockerDrainTest(unittest.IsolatedAsyncioTestCase): finally: store.close() + async def test_foreign_failed_laguna_locator_is_not_resumed(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) / "current" + workspace.mkdir() + (workspace / ".git").mkdir() + directory = workspace / "agent-task" / "group" / "01_task" + directory.mkdir(parents=True) + task = TaskStageTest().make_task(directory) + task.name = "group/01_task" + task.index = 1 + task.plan_hash = "foreign-laguna" + + foreign_attempt = Path(temporary) / "foreign-attempt" + foreign_attempt.mkdir() + foreign_native = foreign_attempt / "session.jsonl" + foreign_native.write_text("{}\n", encoding="utf-8") + foreign_locator = foreign_attempt / "locator.json" + foreign_locator.write_text( + json.dumps( + { + "status": "failed", + "workspace": str((Path(temporary) / "foreign").resolve()), + "workspace_id": "foreign-workspace", + "cli": "pi", + "model": "laguna-s:2.1", + "failure_class": "session-stall", + "native_session_path": str(foreign_native), + } + ), + encoding="utf-8", + ) + observed_resume_locators: list[Path | None] = [] + + async def fake_worker( + workspace_path, + state_store, + selected_task, + *args, + **kwargs, + ): + observed_resume_locators.append(kwargs.get("resume_locator")) + state_store.update_task(selected_task, blocked="test-stop") + return None + + args = SimpleNamespace( + task_group="group", + retry_blocked=False, + dry_run=False, + ) + store = dispatch.StateStore(workspace) + store.mark_active(task, "worker", foreign_locator) + try: + with ( + mock.patch.object( + dispatch, + "scan_tasks", + return_value=[task], + ), + mock.patch.object( + dispatch, + "run_worker", + new=fake_worker, + ), + ): + result = await dispatch.dispatch_with_store( + args, + workspace, + store, + ) + self.assertEqual(result, 2) + self.assertEqual(observed_resume_locators, [None]) + finally: + store.close() + class ReviewSchedulingTest(unittest.TestCase): def test_all_ready_reviews_are_selected_without_numeric_cap(self): - reviews = [ - (mock.sentinel.review_a, "review"), - (mock.sentinel.review_b, "review"), - (mock.sentinel.review_c, "review"), - ] - worker = (mock.sentinel.worker, "worker") - selected, deferred, reason = dispatch.select_dispatch_candidates( - [*reviews, worker] - ) - self.assertEqual(selected, [*reviews, worker]) - self.assertEqual(deferred, []) - self.assertEqual(reason, "") + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + (workspace / ".git").mkdir() + tasks = [ + dispatch.Task( + name=f"group/{index:02d}_task", + directory=workspace / f"task-{index}", + plan=None, + review=None, + user_review=None, + recovery=False, + write_set={ + str((workspace / "src" / f"task-{index}.py").resolve()) + }, + write_set_known=True, + plan_hash=f"hash-{index}", + ) + for index in range(4) + ] + ready = [ + (tasks[0], "review"), + (tasks[1], "review"), + (tasks[2], "review"), + (tasks[3], "worker"), + ] + store = dispatch.StateStore(workspace) + try: + selected, deferred, reason = dispatch.select_dispatch_candidates( + store, + ready, + persist=False, + ) + finally: + store.close() + self.assertEqual(selected, ready) + self.assertEqual(deferred, []) + self.assertEqual(reason, "") def test_new_reviews_join_already_running_review_phase(self): - reviews = [ - (mock.sentinel.review_a, "review"), - (mock.sentinel.review_b, "review"), - ] - worker = (mock.sentinel.worker, "selfcheck") - selected, deferred, reason = dispatch.select_dispatch_candidates( - [*reviews, worker] - ) - self.assertEqual(selected, [*reviews, worker]) - self.assertEqual(deferred, []) - self.assertEqual(reason, "") + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + (workspace / ".git").mkdir() + tasks = [ + dispatch.Task( + name=f"group/{index:02d}_task", + directory=workspace / f"task-{index}", + plan=None, + review=None, + user_review=None, + recovery=False, + write_set={ + str((workspace / "src" / f"task-{index}.py").resolve()) + }, + write_set_known=True, + plan_hash=f"hash-{index}", + ) + for index in range(3) + ] + ready = [ + (tasks[0], "review"), + (tasks[1], "review"), + (tasks[2], "selfcheck"), + ] + store = dispatch.StateStore(workspace) + try: + selected, deferred, reason = dispatch.select_dispatch_candidates( + store, + ready, + persist=False, + ) + finally: + store.close() + self.assertEqual(selected, ready) + self.assertEqual(deferred, []) + self.assertEqual(reason, "") def test_only_declared_same_group_live_predecessors_delay_task(self): task = dispatch.Task( @@ -4692,6 +4916,25 @@ class ReviewSchedulingTest(unittest.TestCase): class WriteSetTest(unittest.TestCase): + def make_claim_task( + self, + workspace: Path, + name: str, + *paths: Path, + plan_hash: str = "plan-0", + ) -> dispatch.Task: + return dispatch.Task( + name=name, + directory=workspace / "agent-task" / name, + plan=None, + review=None, + user_review=None, + recovery=False, + write_set={str(path.resolve()) for path in paths}, + write_set_known=True, + plan_hash=plan_hash, + ) + def test_normalizes_relative_and_absolute_aliases(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) @@ -4751,7 +4994,7 @@ class WriteSetTest(unittest.TestCase): ) self.assertEqual(scanned.errors, []) - def test_recovery_without_matching_plan_does_not_block_dispatch(self): + def test_recovery_without_matching_plan_fails_closed(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) task = workspace / "agent-task" / "recovery" @@ -4764,7 +5007,10 @@ class WriteSetTest(unittest.TestCase): ) [scanned] = dispatch.scan_tasks(workspace, None) self.assertFalse(scanned.write_set_known) - self.assertEqual(scanned.errors, []) + self.assertEqual( + scanned.errors, + ["PLAN Modified Files Summary가 없거나 비어 있거나 유효하지 않다"], + ) def test_rejects_broad_or_outside_workspace_write_sets(self): with tempfile.TemporaryDirectory() as temporary: @@ -4782,6 +5028,255 @@ class WriteSetTest(unittest.TestCase): self.assertFalse(known) self.assertEqual(write_set, set()) + def test_active_plan_without_valid_modified_files_summary_fails_closed(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + directory = workspace / "agent-task" / "missing-write-set" + directory.mkdir(parents=True) + header = "\n" + (directory / "PLAN-local-G05.md").write_text( + header + "## Background\n\nNo file table.\n", + encoding="utf-8", + ) + (directory / "CODE_REVIEW-local-G05.md").write_text( + header, + encoding="utf-8", + ) + + [task] = dispatch.scan_tasks(workspace, None) + + self.assertFalse(task.write_set_known) + self.assertEqual( + task.errors, + ["PLAN Modified Files Summary가 없거나 비어 있거나 유효하지 않다"], + ) + + def test_workspace_claims_persist_replace_wait_and_release_on_completion(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + (workspace / ".git").mkdir() + shared = workspace / "src" / "shared.py" + disjoint = workspace / "src" / "disjoint.py" + expansion = workspace / "src" / "expansion.py" + alpha = self.make_claim_task( + workspace, + "alpha/01_task", + shared, + ) + beta = self.make_claim_task( + workspace, + "beta/01_task", + shared, + ) + gamma = self.make_claim_task( + workspace, + "gamma/01_task", + disjoint, + ) + + store = dispatch.StateStore(workspace) + try: + selected, deferred, _ = dispatch.select_dispatch_candidates( + store, + [(alpha, "worker"), (beta, "worker"), (gamma, "review")], + persist=True, + ) + self.assertEqual(selected, [(gamma, "review"), (alpha, "worker")]) + self.assertEqual( + deferred, + [ + ( + beta, + "worker", + "write claim 충돌 대기: " + f"owner=alpha/01_task; path={shared.resolve()}", + ) + ], + ) + alpha_acquired_at = store.data["write_claims"][alpha.name][ + "acquired_at" + ] + finally: + store.close() + + reopened = dispatch.StateStore(workspace) + try: + self.assertEqual( + set(reopened.data["write_claims"]), + {alpha.name, gamma.name}, + ) + alpha_followup = self.make_claim_task( + workspace, + alpha.name, + shared, + expansion, + plan_hash="plan-1", + ) + selected, deferred, _ = dispatch.select_dispatch_candidates( + reopened, + [(alpha_followup, "worker")], + persist=True, + ) + self.assertEqual(selected, [(alpha_followup, "worker")]) + self.assertEqual(deferred, []) + self.assertEqual( + reopened.data["write_claims"][alpha.name]["acquired_at"], + alpha_acquired_at, + ) + self.assertEqual( + reopened.data["write_claims"][alpha.name]["paths"], + sorted([str(shared.resolve()), str(expansion.resolve())]), + ) + + conflicting_followup = self.make_claim_task( + workspace, + alpha.name, + shared, + disjoint, + plan_hash="plan-2", + ) + selected, deferred, _ = dispatch.select_dispatch_candidates( + reopened, + [(conflicting_followup, "worker")], + persist=True, + ) + self.assertEqual(selected, []) + self.assertIn(f"owner={gamma.name}", deferred[0][2]) + self.assertEqual( + reopened.data["write_claims"][alpha.name]["plan_hash"], + "plan-1", + ) + + archive = workspace / "archive-alpha" + archive.mkdir() + (archive / "complete.log").write_text( + "complete\n", + encoding="utf-8", + ) + reopened.mark_orchestration_task_complete( + "alpha", + alpha.name, + archive, + ) + self.assertNotIn(alpha.name, reopened.data["write_claims"]) + + selected, deferred, _ = dispatch.select_dispatch_candidates( + reopened, + [(beta, "worker")], + persist=True, + ) + self.assertEqual(selected, [(beta, "worker")]) + self.assertEqual(deferred, []) + finally: + reopened.close() + + def test_unknown_write_set_cannot_acquire_claim(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + (workspace / ".git").mkdir() + task = self.make_claim_task( + workspace, + "group/01_unknown", + workspace / "src" / "unknown.py", + ) + task.write_set_known = False + task.write_set = set() + store = dispatch.StateStore(workspace) + try: + selected, deferred, _ = dispatch.select_dispatch_candidates( + store, + [(task, "worker")], + persist=True, + ) + self.assertEqual(selected, []) + self.assertIn("valid non-empty", deferred[0][2]) + self.assertEqual(store.data["write_claims"], {}) + finally: + store.close() + + def test_claim_preview_is_stateless(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + (workspace / ".git").mkdir() + task = self.make_claim_task( + workspace, + "group/01_preview", + workspace / "src" / "preview.py", + ) + store = dispatch.StateStore(workspace) + try: + before = json.loads(json.dumps(store.data)) + selected, deferred, _ = dispatch.select_dispatch_candidates( + store, + [(task, "worker")], + persist=False, + ) + self.assertEqual(selected, [(task, "worker")]) + self.assertEqual(deferred, []) + self.assertEqual(store.data, before) + self.assertFalse(store.path.exists()) + finally: + store.close() + + def test_active_legacy_task_adopts_exclusive_workspace_claim(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + (workspace / ".git").mkdir() + legacy = self.make_claim_task( + workspace, + "legacy/01_active", + workspace / "src" / "unknown.py", + ) + legacy.write_set_known = False + legacy.write_set = set() + candidate = self.make_claim_task( + workspace, + "other/01_candidate", + workspace / "src" / "disjoint.py", + ) + store = dispatch.StateStore(workspace) + try: + store.adopt_active_write_claim(legacy) + claim = store.data["write_claims"][legacy.name] + self.assertTrue(claim["exclusive"]) + self.assertEqual(claim["paths"], []) + + selected, deferred, _ = dispatch.select_dispatch_candidates( + store, + [(candidate, "worker")], + persist=True, + ) + self.assertEqual(selected, []) + self.assertIn(f"owner={legacy.name}", deferred[0][2]) + self.assertIn("", deferred[0][2]) + finally: + store.close() + + def test_state_workspace_identity_is_persisted_and_validated(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + (workspace / ".git").mkdir() + store = dispatch.StateStore(workspace) + state_path = store.path + expected_id = store.workspace_id + try: + store.save() + finally: + store.close() + + state = json.loads(state_path.read_text(encoding="utf-8")) + self.assertEqual( + state["workspace_identity"], + {"id": expected_id, "root": str(workspace.resolve())}, + ) + state["workspace_identity"]["root"] = str( + (workspace / "foreign").resolve() + ) + state_path.write_text(json.dumps(state), encoding="utf-8") + + with self.assertRaises(dispatch.DispatcherTerminalStateError): + dispatch.StateStore(workspace) + def test_review_progress_signature_ignores_dispatcher_work_log(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) @@ -5138,25 +5633,257 @@ class WorkLogArchiveTest(unittest.TestCase): process.terminate() process.wait(timeout=5) + def test_workspace_bound_liveness_rejects_foreign_and_accepts_current_locators(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) / "current" + workspace.mkdir() + (workspace / ".git").mkdir() + store = dispatch.StateStore(workspace) + try: + foreign_attempt = Path(temporary) / "foreign" / "attempt" + foreign_attempt.mkdir(parents=True) + foreign_locator = foreign_attempt / "locator.json" + foreign_locator.write_text( + json.dumps( + { + "status": "running", + "workspace": str((Path(temporary) / "foreign").resolve()), + "workspace_id": "foreign-workspace", + "agent_pid": os.getpid(), + } + ), + encoding="utf-8", + ) + with mock.patch.object( + dispatch, + "process_is_alive", + side_effect=AssertionError( + "foreign locator must be rejected before PID inspection" + ), + ): + live, detail = dispatch.external_active_is_live( + {"active_locator": str(foreign_locator)}, + expected_workspace=store.workspace, + expected_workspace_id=store.workspace_id, + expected_runs_root=store.runs, + ) + self.assertFalse(live) + self.assertIn("foreign workspace locator path", detail) + + current_attempt = store.runs / "current-attempt" + current_attempt.mkdir() + current_locator = current_attempt / "locator.json" + current_locator.write_text( + json.dumps( + { + "status": "running", + "workspace": str(store.workspace), + "workspace_id": store.workspace_id, + "agent_pid": os.getpid(), + } + ), + encoding="utf-8", + ) + live, detail = dispatch.external_active_is_live( + {"active_locator": str(current_locator)}, + expected_workspace=store.workspace, + expected_workspace_id=store.workspace_id, + expected_runs_root=store.runs, + ) + self.assertTrue(live) + self.assertIn("agent_pid", detail) + + legacy_attempt = store.runs / "legacy-attempt" + legacy_attempt.mkdir() + legacy_locator = legacy_attempt / "locator.json" + legacy_locator.write_text( + json.dumps( + { + "status": "running", + "agent_pid": os.getpid(), + } + ), + encoding="utf-8", + ) + live, detail = dispatch.external_active_is_live( + {"active_locator": str(legacy_locator)}, + expected_workspace=store.workspace, + expected_workspace_id=store.workspace_id, + expected_runs_root=store.runs, + ) + self.assertTrue(live) + self.assertIn("agent_pid", detail) + + foreign_stream = foreign_attempt / "stream.log" + foreign_stream.write_text("foreign output\n", encoding="utf-8") + legacy_locator.write_text( + json.dumps( + { + "status": "running", + "agent_pid": os.getpid(), + "stream_log": str(foreign_stream), + } + ), + encoding="utf-8", + ) + with mock.patch.object( + dispatch, + "process_is_alive", + side_effect=AssertionError( + "foreign stream evidence must fail before PID inspection" + ), + ): + live, detail = dispatch.external_active_is_live( + {"active_locator": str(legacy_locator)}, + expected_workspace=store.workspace, + expected_workspace_id=store.workspace_id, + expected_runs_root=store.runs, + ) + self.assertFalse(live) + self.assertIn("foreign workspace locator evidence", detail) + finally: + store.close() + + def test_workspace_bound_liveness_rejects_mismatched_identity_inside_runs(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) + (workspace / ".git").mkdir() + store = dispatch.StateStore(workspace) + try: + attempt = store.runs / "foreign-identity" + attempt.mkdir() + locator = attempt / "locator.json" + locator.write_text( + json.dumps( + { + "status": "running", + "workspace": str((workspace / "other").resolve()), + "workspace_id": "foreign-workspace", + "agent_pid": os.getpid(), + } + ), + encoding="utf-8", + ) + with mock.patch.object( + dispatch, + "process_is_alive", + side_effect=AssertionError( + "mismatched workspace must fail before PID inspection" + ), + ): + live, detail = dispatch.external_active_is_live( + {"active_locator": str(locator)}, + expected_workspace=store.workspace, + expected_workspace_id=store.workspace_id, + expected_runs_root=store.runs, + ) + self.assertFalse(live) + self.assertIn("foreign workspace locator id", detail) + finally: + store.close() + + def test_workspace_bound_laguna_resume_rejects_foreign_locator_and_native_session(self): + with tempfile.TemporaryDirectory() as temporary: + workspace = Path(temporary) / "current" + workspace.mkdir() + (workspace / ".git").mkdir() + store = dispatch.StateStore(workspace) + try: + foreign_attempt = Path(temporary) / "foreign-attempt" + foreign_attempt.mkdir() + foreign_native = foreign_attempt / "session.jsonl" + foreign_native.write_text("{}\n", encoding="utf-8") + foreign_locator = foreign_attempt / "locator.json" + foreign_locator.write_text( + json.dumps( + { + "status": "failed", + "workspace": str((Path(temporary) / "foreign").resolve()), + "workspace_id": "foreign-workspace", + "cli": "pi", + "model": "laguna-s:2.1", + "failure_class": "session-stall", + "native_session_path": str(foreign_native), + } + ), + encoding="utf-8", + ) + self.assertIsNone( + dispatch.laguna_resume_locator( + {"active_locator": str(foreign_locator)}, + expected_workspace=store.workspace, + expected_workspace_id=store.workspace_id, + expected_runs_root=store.runs, + ) + ) + + current_attempt = store.runs / "current-laguna" + current_attempt.mkdir() + current_native = current_attempt / "session.jsonl" + current_native.write_text("{}\n", encoding="utf-8") + current_locator = current_attempt / "locator.json" + record = { + "status": "failed", + "workspace": str(store.workspace), + "workspace_id": store.workspace_id, + "cli": "pi", + "model": "laguna-s:2.1", + "failure_class": "session-stall", + "native_session_path": str(current_native), + } + current_locator.write_text( + json.dumps(record), + encoding="utf-8", + ) + self.assertEqual( + dispatch.laguna_resume_locator( + {"active_locator": str(current_locator)}, + expected_workspace=store.workspace, + expected_workspace_id=store.workspace_id, + expected_runs_root=store.runs, + ), + current_locator, + ) + + record["native_session_path"] = str(foreign_native) + current_locator.write_text( + json.dumps(record), + encoding="utf-8", + ) + self.assertIsNone( + dispatch.laguna_resume_locator( + {"active_locator": str(current_locator)}, + expected_workspace=store.workspace, + expected_workspace_id=store.workspace_id, + expected_runs_root=store.runs, + ) + ) + finally: + store.close() + def test_orchestration_keeps_pidless_stream_evidence_active(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() - stream = workspace / "stream.log" - stream.write_text("reasoning\n", encoding="utf-8") - locator = workspace / "locator.json" - locator.write_text( - json.dumps( - { - "status": "running", - "cli": "codex", - "stream_log": str(stream), - } - ), - encoding="utf-8", - ) store = dispatch.StateStore(workspace) try: + attempt = store.runs / "attempt" + attempt.mkdir() + stream = attempt / "stream.log" + stream.write_text("reasoning\n", encoding="utf-8") + locator = attempt / "locator.json" + locator.write_text( + json.dumps( + { + "status": "running", + "workspace": str(workspace.resolve()), + "workspace_id": store.workspace_id, + "cli": "codex", + "stream_log": str(stream), + } + ), + encoding="utf-8", + ) store.data["orchestrations"] = { "group": { "status": "running", @@ -6425,18 +7152,21 @@ class DispatcherConvergenceSimulationTest(unittest.IsolatedAsyncioTestCase): work_log.write_text("final timeline\n", encoding="utf-8") active = {"worker": set(), "selfcheck": set(), "review": set()} + active_tasks: set[str] = set() maximum = {"worker": 0, "selfcheck": 0, "review": 0} overlap_violations: list[tuple[str, set[str]]] = [] review_attempts: dict[str, int] = {} def enter(stage: str, task_name: str) -> None: active[stage].add(task_name) + active_tasks.add(task_name) maximum[stage] = max(maximum[stage], len(active[stage])) - if {"sim/01_alpha", "sim/04_conflict"} <= active[stage]: - overlap_violations.append((stage, set(active[stage]))) + if {"sim/01_alpha", "sim/04_conflict"} <= active_tasks: + overlap_violations.append((stage, set(active_tasks))) def leave(stage: str, task_name: str) -> None: active[stage].remove(task_name) + active_tasks.remove(task_name) async def fake_worker(workspace_path, store, task, *args, **kwargs): enter("worker", task.name) @@ -6550,10 +7280,7 @@ class DispatcherConvergenceSimulationTest(unittest.IsolatedAsyncioTestCase): self.assertGreaterEqual(maximum["worker"], 2) self.assertGreaterEqual(maximum["selfcheck"], 2) self.assertGreaterEqual(maximum["review"], 2) - self.assertEqual( - {stage for stage, _ in overlap_violations}, - {"worker", "selfcheck", "review"}, - ) + self.assertEqual(overlap_violations, []) self.assertGreater(scan_tasks.call_count, 1) self.assertLessEqual(scan_tasks.call_count, 5) self.assertTrue( @@ -6587,7 +7314,13 @@ class DynamicFailoverBudgetTest(unittest.TestCase): directory = workspace / "agent-task" / "budget/01_unit" directory.mkdir(parents=True) header = "\n" - (directory / "PLAN-local-G07.md").write_text(header, encoding="utf-8") + (directory / "PLAN-local-G07.md").write_text( + header + + "## Modified Files Summary\n\n" + "| File | Item |\n|---|---|\n" + "| `src/budget.py` | TEST-1 |\n", + encoding="utf-8", + ) (directory / "CODE_REVIEW-local-G07.md").write_text(header, encoding="utf-8") return dispatch.scan_tasks(workspace, None)[0] @@ -6785,7 +7518,13 @@ class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCas directory = workspace / "agent-task" / "failover/01_unit" directory.mkdir(parents=True, exist_ok=True) header = "\n" - (directory / f"PLAN-{lane}-G{grade:02d}.md").write_text(header, encoding="utf-8") + (directory / f"PLAN-{lane}-G{grade:02d}.md").write_text( + header + + "## Modified Files Summary\n\n" + "| File | Item |\n|---|---|\n" + "| `src/failover.py` | TEST-1 |\n", + encoding="utf-8", + ) (directory / f"CODE_REVIEW-{lane}-G{grade:02d}.md").write_text(header, encoding="utf-8") tasks = dispatch.scan_tasks(workspace, None) return tasks[0] @@ -7526,7 +8265,13 @@ class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase): directory = workspace / "agent-task" / "selector_dispatch_integration" / unit directory.mkdir(parents=True, exist_ok=True) header = f"\n" - (directory / f"PLAN-{lane}-G{grade:02d}.md").write_text(header, encoding="utf-8") + (directory / f"PLAN-{lane}-G{grade:02d}.md").write_text( + header + + "## Modified Files Summary\n\n" + "| File | Item |\n|---|---|\n" + f"| `src/{unit}.py` | TEST-1 |\n", + encoding="utf-8", + ) (directory / f"CODE_REVIEW-{lane}-G{grade:02d}.md").write_text(header, encoding="utf-8") tasks = dispatch.scan_tasks(workspace, None) for task in tasks: @@ -8368,7 +9113,7 @@ class ThroughputQuotaBatchTest(unittest.TestCase): (directory / f"PLAN-{lane}-G{grade:02d}.md").write_text( header + "## 수정 파일 요약\n\n| 파일 | 항목 |\n|---|---|\n" - + "| src/route.py | ROUTE-1 |\n", + + f"| `src/{name.replace('/', '_')}.py` | ROUTE-1 |\n", encoding="utf-8", ) (directory / f"CODE_REVIEW-{lane}-G{grade:02d}.md").write_text( @@ -8580,7 +9325,7 @@ class ThroughputQuotaBatchTest(unittest.TestCase): locator.write_text(json.dumps(record), encoding="utf-8") return locator - def test_same_target_tasks_unbound_admission_without_cap(self): + def test_same_provider_target_tasks_with_disjoint_write_sets_admit_without_cap(self): async def run(): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) @@ -9846,12 +10591,16 @@ class ThroughputQuotaBatchTest(unittest.TestCase): # Pre-populate: simulate that the first attempt already committed # a handoff and created an active locator with a live agent PID. - first_attempt_dir = workspace / "20260726T230000SZ__retry-worker-01" + first_attempt_dir = ( + store.runs / "20260726T230000SZ__retry-worker-01" + ) first_attempt_dir.mkdir(parents=True) first_locator_path = first_attempt_dir / "locator.json" fake_agent_pid = 99999 first_locator_data = { "status": "running", + "workspace": str(workspace.resolve()), + "workspace_id": store.workspace_id, "task": t_task.name, "role": "worker", "attempt": 0, @@ -9859,7 +10608,9 @@ class ThroughputQuotaBatchTest(unittest.TestCase): "agent_process_start_token": "fake-token-abc123", "dispatcher_pid": os.getpid(), "dispatcher_process_start_token": "fake-dispatcher-token", - "agent_process_marker": f"retry-worker-01__{uuid.uuid4()}", + "agent_process_marker": ( + f"w{store.workspace_id}__retry-worker-01__{uuid.uuid4()}" + ), "retry_handoff_id": "stable-handoff-id-restart-001", "cli": "agy", "model": "Gemini 3.6 Flash (Medium)", diff --git a/agent-task/archive/2026/07/dispatcher_workspace_ownership/code_review_cloud_G08_0.log b/agent-task/archive/2026/07/dispatcher_workspace_ownership/code_review_cloud_G08_0.log new file mode 100644 index 0000000..296294a --- /dev/null +++ b/agent-task/archive/2026/07/dispatcher_workspace_ownership/code_review_cloud_G08_0.log @@ -0,0 +1,183 @@ + + +# Code Review Reference - REFACTOR + +> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.** +> The task is NOT complete until every implementation-owned section below is filled in. +> Complete the `Implementation Checklist`; the final checklist item is mandatory before saving. +> Fill implementation-owned sections, then stop with active files in place and report ready for review. +> If implementation is blocked, record the exact blocker, attempted commands/output, and resume condition only in implementation-owned evidence fields. +> Do not ask the user directly, present choices, call user-input tools, create control-plane stop files, or classify the next state. +> Finalization (`Code Review Result`, log rename, `complete.log`, archive moves, `Review-Only Checklist`) is review-agent-only, even after compaction/resume. +> Follow the ownership table at the bottom of this file for which sections you own. + +## Overview + +date=2026-07-29 +task=dispatcher_workspace_ownership, plan=0, tag=REFACTOR + +## For the Review Agent + +> **[REVIEW AGENT ONLY]** The finalization steps below are review-agent only. Implementing agents must not execute this section. + +Compare implementation of each item against source files and verify that output in `Verification Results` matches code. +Review completion means the following steps are finished: + +1. Append verdict and `review_rework_count` / `evidence_integrity_failure` routing signals. +2. Archive `CODE_REVIEW-cloud-G08.md` → `code_review_cloud_G08_0.log` and `PLAN-cloud-G08.md` → `plan_cloud_G08_0.log`. +3. If PASS, write `complete.log` and move active task directory to `agent-task/archive/YYYY/MM/dispatcher_workspace_ownership/`. If WARN/FAIL, fully write the next filesystem state required by the code-review skill. +4. If PASS and task group is `m-`, report completion event metadata. Roadmap state check and `update-roadmap` calls are runtime responsibilities. +5. Check applicable `Review-Only Checklist` items at the final `.log` location before reporting. + +--- + +## Implementation Item Completion + +| Item | Status | +|------|---------| +| REFACTOR-1 | [x] | +| REFACTOR-2 | [x] | +| REFACTOR-3 | [x] | + +## Implementation Checklist + +- [x] Bind dispatcher state, locator records, process markers, and liveness recovery to canonical physical workspace identity. +- [x] Enforce persistent Modified Files Summary claims before worker/selfcheck/review admission, wait on collisions, retain across follow-ups/restarts, and release only after verified completion. +- [x] Update project dispatcher policy and mocked regression/convergence tests for workspace isolation, fail-closed write sets, claim lifecycle, and safe parallelism. +- [x] Run deterministic syntax and full dispatcher unit verification. +- [x] Fill implementation-owned sections in CODE_REVIEW-*-G??.md with actual implementation notes and verification output. + +## Review-Only Checklist + +> **[REVIEW AGENT ONLY]** This checklist is used only by the review agent. +> Implementing agents must not modify or check this section. + +- [x] Append one verdict of `PASS`, `WARN`, or `FAIL` and verified `review_rework_count`, `evidence_integrity_failure` to `Code Review Result`. +- [x] Verify that verdict, `Dimension Assessment`, and Required/Suggested/Nit classifications match. +- [x] Archive active `CODE_REVIEW-*-G??.md` to `code_review_cloud_G08_0.log`. +- [x] Archive active `PLAN-*-G??.md` to `plan_cloud_G08_0.log`. +- [x] Verify that the Agent-Ops managed block in `.gitignore` unignores `agent-task/**/*.md` and `agent-task/**/*.log` and ignores `agent-roadmap/current.md`. +- [x] If PASS, write `complete.log` based on `agent-ops/skills/common/code-review/templates/complete-log-template.md` and leave no active `.md` files. +- [x] If PASS, move active task directory `agent-task/dispatcher_workspace_ownership/` to `agent-task/archive/YYYY/MM/dispatcher_workspace_ownership/` and update this checklist at the final archive path. +- [ ] If PASS and task group is `m-`, report completion event metadata for runtime, without modifying roadmap or directly calling `update-roadmap`. +- [ ] If PASS for split work, remove empty active parent `agent-task/{task_name}/` or verify it was kept due to remaining siblings/files. +- [ ] If WARN/FAIL, write the next filesystem state matching code-review verdict and do not write `complete.log`. + +## Deviations from Plan + +No implementation-scope or verification-command deviation. During review, the original workspace-ownership criterion exposed two additional resume/evidence seams, so those seams and their regressions were corrected within REFACTOR-1. The planned unit command was run exactly after the review fixes. Because its lifecycle-fixture output is verbose, the invocation used shell redirection to preserve complete output at `/tmp/iop-dispatcher-workspace-ownership-review-unittest.log`. + +## Key Design Decisions + +- State records, locator paths, locator metadata, and process markers are bound to the canonical physical workspace id before liveness evidence is accepted. +- Laguna recovery and Pi invocation independently revalidate locator, stream, and native-session ownership at their final resume seams, so a foreign failed locator cannot bypass the liveness gate. +- The write-claim ledger is global to one physical workspace rather than one task group. Ordered admission updates it atomically under the existing dispatcher lock. +- A task owns its claim through all execution stages and follow-up generations. A conflicting expansion keeps the prior claim unchanged, while verified archive completion releases it. +- Active legacy work without a valid file list adopts an exclusive workspace claim, preventing unsafe guesses until the task is repaired or verified complete. +- Missing, duplicate, broad, directory, wildcard, outside-workspace, or otherwise invalid modified-file tables fail closed; dry-run previews admission without mutating state. + +## Reviewer Checkpoints + +- Reject foreign-workspace locators before consulting PID, process-marker, native-session, or stream evidence. +- Verify that claim acquisition is atomic per physical workspace and collision reports identify both owner and canonical path. +- Verify that a claim survives worker, self-check, review, retry, restart, and compatible follow-up transitions, then releases only after verified completion. +- Verify that disjoint tasks still run in parallel and separate physical worktrees/clones use independent state and claim ledgers. +- Verify that all provider-facing paths remain mocked in tests. + +## Verification Results + +### Python bytecode compilation + +Command: + +```bash +python3 -m py_compile agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py +``` + +Actual output: + +```text +Process exited with code 0 +stdout: +stderr: +``` + +### Dispatcher unit suite + +Command: + +```bash +python3 -m unittest discover -s agent-ops/skills/project/orchestrate-agent-task-loop/tests -p 'test_dispatch.py' +``` + +Actual output: + +The planned command exited `0`. Its captured result ended with: + +```text +---------------------------------------------------------------------- +Ran 219 tests in 13.584s + +OK +``` + +The complete capture is `/tmp/iop-dispatcher-workspace-ownership-review-unittest.log`. Capture command and actual tail: + +```text +$ python3 -m unittest discover -s agent-ops/skills/project/orchestrate-agent-task-loop/tests -p 'test_dispatch.py' > /tmp/iop-dispatcher-workspace-ownership-review-unittest.log 2>&1 +---------------------------------------------------------------------- +Ran 219 tests in 13.584s + +OK +``` + +--- + +> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section?** +> If anything is blank, go back and fill it in before saving this file. +> Leave review-agent-only sections unchanged. + +## Section Ownership + +| Section | Owner | Note | +|---------|-------|------| +| Header comment, Overview, Review Agent Instructions | Fixed at stub creation | Implementing agent must not modify or execute these (archive, complete.log, and task-directory archive move are review-agent only) | +| Roadmap Targets | Fixed at stub creation from plan when present | Implementing agent must not modify; code-review copies it into `complete.log` as `Roadmap Completion` only on PASS | +| Archive Evidence Snapshot | Fixed at stub creation from plan when present | Implementing agent uses it as default prior-loop context; read only the specific archive files cited there when more detail is required | +| Agent UI Completion | Mixed | Present only for plan-required agent-ui code work; implementing agent fills actual evidence, review agent applies `구현됨` status/evidence update on PASS and copies the section into `complete.log` | +| Implementation Item Completion (item names) | Fixed at stub creation | Implementing agent checks `[ ]` → `[x]` only | +| Implementation Checklist (item text/order) | Fixed at stub creation from plan | Implementing agent checks `[ ]` → `[x]` only | +| Review-Only Checklist | Review agent only | Implementing agent must not modify or check this section | +| Deviations from Plan, Key Design Decisions | Implementing agent | Replace placeholder text with actual content | +| Reviewer Checkpoints | Fixed at stub creation | Pre-filled from plan | +| Verification Results (section headings + commands) | Fixed at stub creation | Implementing agent fills in command output only; command changes require a `Deviations from Plan` entry | +| Code Review Result | Review agent appends | Not included in stub | + +## Code Review Result + +### Overall Verdict + +PASS + +### Dimension Assessment + +- Correctness: Pass — workspace-owned liveness, recovery, final resume seams, and persistent write-claim admission match the plan's invariants. +- Completeness: Pass — all three implementation items and their required policy/test updates are present. +- Test coverage: Pass — foreign/current/legacy locator ownership, Laguna/Pi resume boundaries, claim collision/lifecycle, restart behavior, and disjoint parallelism are covered. +- API contract: Pass — no public symbol was removed, and the project dispatcher contract now describes the enforced runtime behavior. +- Code quality: Pass — ownership and claim validation are centralized, fail closed, and integrated through existing state and scheduler boundaries. +- Implementation deviation: Pass — review fixes remained within REFACTOR-1 and no out-of-scope product, common-rule, or roadmap files changed. +- Verification trust: Pass — bytecode compilation and all 219 dispatcher tests passed from the reviewed working tree. + +### Findings + +None. + +### Routing Signals + +- `review_rework_count=0` +- `evidence_integrity_failure=false` + +### Next Step + +Archive the reviewed pair, write `complete.log`, and move the completed task to its dated archive. diff --git a/agent-task/archive/2026/07/dispatcher_workspace_ownership/complete.log b/agent-task/archive/2026/07/dispatcher_workspace_ownership/complete.log new file mode 100644 index 0000000..f300a44 --- /dev/null +++ b/agent-task/archive/2026/07/dispatcher_workspace_ownership/complete.log @@ -0,0 +1,34 @@ +# Complete - dispatcher_workspace_ownership + +## 완료 일시 + +2026-07-29 + +## 요약 + +물리 워크스페이스별 실행 소유권과 지속 파일 쓰기 잠금을 dispatcher에 적용했으며, Plan 1회와 공식 리뷰 1회로 최종 PASS했다. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_cloud_G08_0.log` | `code_review_cloud_G08_0.log` | PASS | 외부 워크스페이스 locator/재개 차단과 동일 워크스페이스 파일 충돌 직렬화를 구현하고 검증했다. | + +## 구현/정리 내용 + +- dispatcher 상태, locator, 프로세스 마커, liveness 및 resume 증거를 canonical 물리 워크스페이스에 귀속시켰다. +- PLAN의 canonical 수정 파일을 원자적으로 claim하고 충돌 작업은 대기시키며, 완료 아카이브 검증 후에만 claim을 해제하도록 했다. +- 별도 worktree/clone은 독립 실행하고 동일 워크스페이스의 교차 task-group 충돌은 직렬화하도록 프로젝트 정책과 테스트를 갱신했다. + +## 최종 검증 + +- `python3 -m py_compile agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py` - PASS; exit code 0, 출력 없음. +- `python3 -m unittest discover -s agent-ops/skills/project/orchestrate-agent-task-loop/tests -p 'test_dispatch.py'` - PASS; 219 tests, `OK`; 전체 출력=`/tmp/iop-dispatcher-workspace-ownership-review-unittest.log`. + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/archive/2026/07/dispatcher_workspace_ownership/plan_cloud_G08_0.log b/agent-task/archive/2026/07/dispatcher_workspace_ownership/plan_cloud_G08_0.log new file mode 100644 index 0000000..94e20dc --- /dev/null +++ b/agent-task/archive/2026/07/dispatcher_workspace_ownership/plan_cloud_G08_0.log @@ -0,0 +1,219 @@ + + +# Dispatcher Workspace Ownership and Runtime Write Claims + +## For the Implementing Agent + +Filling every implementation-owned section in `CODE_REVIEW-*-G??.md` is mandatory. Run the specified verification, record actual implementation notes and output, keep the active task files in place, and report ready for review; finalization belongs only to the code-review skill. If blocked, record the exact blocker, attempted commands/output, and resume condition only in implementation-owned evidence fields. Do not ask the user, call user-input tools, create control-plane stop files, classify the next state, archive logs, or write `complete.log`. + +## Background + +Persisted dispatcher locators can currently make one physical checkout monitor a session created from another checkout because liveness recovery trusts process and stream evidence without first validating workspace ownership. The dispatcher also parses each PLAN's modified-file table but deliberately ignores it during admission, allowing independent tasks that target the same file to run concurrently. Workspace-bound liveness and persistent per-workspace write claims are required to make isolated worktrees parallel-safe while serializing conflicting work inside one checkout. + +## Analysis + +### Files Read + +- `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py` +- `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py` +- `agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md` +- `agent-test/local/rules.md` +- `agent-test/local/testing-smoke.md` + +### SDD Criteria + +Not applicable. This is a non-roadmap dispatcher runtime correction and has no Milestone-linked SDD gate. + +### Test Environment Rules + +`test_env=local`. `agent-test/local/rules.md` and the matched `agent-test/local/testing-smoke.md` profile were present and read. The profile's Edge/Node product-runtime commands do not target this Python dispatcher, so deterministic fallback verification uses the existing dispatcher `unittest` suite plus Python bytecode compilation. No verification leaves the current checkout, no real provider invocation is permitted, and test-rule maintenance is not needed. + +### Test Coverage Gaps + +- Existing tests cover locator PID/marker/stream liveness, but do not reject a live locator owned by another physical workspace. +- Existing write-set tests cover extraction and canonicalization, but do not require a valid non-empty table for admission. +- Existing scheduler tests explicitly expect overlapping write sets to run concurrently; they must be revised to prove collision waiting, claim persistence, follow-up retention, verified-completion release, and continued parallelism for disjoint paths. +- Existing convergence simulation must be updated so disjoint tasks overlap while tasks targeting the same file serialize through worker, self-check, and review. + +### Symbol References + +No public symbols are renamed or removed. Internal call sites of `external_active_is_live` are `orchestration_live_agent_processes` and the main scheduler's persisted-active-stage recovery path. Internal call sites of `select_dispatch_candidates` are in the scheduler and unit tests. + +### Split Judgment + +Use one plan. Workspace identity validation, claim acquisition, claim retention across stages/restarts, and claim release after verified completion form one indivisible execution-ownership invariant; landing only part of it would either permit duplicate monitoring or leave unsafe write concurrency. + +### Scope Rationale + +Modify only the project dispatcher script, its project policy skill, and its tests. Do not modify centrally managed common rules/skills, roadmap files, product Edge/Node runtime, or establish a repository-global lock shared by separate worktrees/clones. Claims are intentionally scoped to one canonical physical workspace so isolated checkouts may execute independently. + +### Final Routing + +- `evaluation_mode`: `pair` +- `finalizer`: `local-fit` +- Build target: closure `risk-boundary`, grade `G08`, route `cloud-G08` +- Review target: closure `official-review`, grade `G08`, route `cloud-G08` +- `large_indivisible_context`: `false` +- Positive loop risks: `temporal_state`, `concurrent_consistency`, `boundary_contract`, `structured_interpretation` (4) +- Recovery signals: `review_rework_count=0`, `evidence_integrity_failure=false` +- Capability-gap evidence: none +- Canonical files: `PLAN-cloud-G08.md`, `CODE_REVIEW-cloud-G08.md` + +## Implementation Checklist + +- [ ] Bind dispatcher state, locator records, process markers, and liveness recovery to canonical physical workspace identity. +- [ ] Enforce persistent Modified Files Summary claims before worker/selfcheck/review admission, wait on collisions, retain across follow-ups/restarts, and release only after verified completion. +- [ ] Update project dispatcher policy and mocked regression/convergence tests for workspace isolation, fail-closed write sets, claim lifecycle, and safe parallelism. +- [ ] Run deterministic syntax and full dispatcher unit verification. +- [ ] Fill implementation-owned sections in CODE_REVIEW-*-G??.md with actual implementation notes and verification output. + +### [REFACTOR-1] Bind runtime liveness to physical workspace identity + +#### Problem + +At `dispatch.py:439-510`, `StateStore` derives a local hash only for a fallback directory and does not persist or expose canonical workspace ownership. At `dispatch.py:2925-2995`, `external_active_is_live` accepts a locator's PID, process marker, and stream evidence before proving that the locator belongs to the current checkout. A stale or copied state record can therefore make this dispatcher monitor a session launched from another worktree or clone. + +#### Solution + +Persist and validate canonical workspace root/id metadata in dispatcher state, include the id in locators, and namespace attempt process markers by workspace id. Require matching locator workspace identity at production liveness call sites before accepting any process or stream evidence; legacy locators are accepted only when their locator path is physically under this store's `runs` root. + +Before (`dispatch.py:2925`): + +```python +def external_active_is_live(state: dict[str, Any]) -> tuple[bool, str]: + raw_locator = state.get("active_locator") +``` + +After: + +```python +def external_active_is_live( + state: dict[str, Any], + *, + expected_workspace: Path | None = None, + expected_workspace_id: str | None = None, + expected_runs_root: Path | None = None, +) -> tuple[bool, str]: + raw_locator = state.get("active_locator") + # Reject foreign workspace identity before PID/marker/stream checks. +``` + +#### Modified Files and Checklist + +- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py`: persist workspace identity, namespace locators/markers, and validate liveness ownership. +- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py`: add matching, foreign, and legacy-locator ownership regressions. + +#### Test Strategy + +Write tests in `test_dispatch.py` proving that a foreign locator with a live current PID is rejected, a matching locator is retained, a legacy locator is accepted only below the current store's runs root, and generated locators/markers contain the current workspace id. All process evidence is mocked or uses the current test process; no provider is started. + +#### Verification + +Run the full dispatcher unit command in `Final Verification`; all workspace ownership tests and pre-existing liveness tests must pass. + +### [REFACTOR-2] Enforce persistent per-workspace write claims + +#### Problem + +At `dispatch.py:950-991`, PLAN write sets are parsed but missing, malformed, wildcard, empty, directory, or outside-workspace targets only result in `write_set_known=False`. At `dispatch.py:5159-5167`, every dependency-ready task is admitted and the implementation explicitly states that write sets never create a cross-task barrier. This permits two stages in the same physical checkout to edit the same canonical file concurrently. + +#### Solution + +Treat an active/recovery PLAN without one valid, non-empty `Modified Files Summary` as a task-local fail-closed error. Persist a workspace-scoped write-claim ledger keyed by task generation; atomically admit ordered ready candidates only when their canonical file set has no intersection with another task claim. Keep the claim across worker, self-check, official review, retries, dispatcher restarts, and follow-up generations; replace/expand a task's own claim only when the new set does not conflict. Report collisions as waiting rather than logical dependency blockers, adopt conservatively exclusive claims for externally active legacy tasks with unknown write sets, and release only after a verified completed archive. + +Before (`dispatch.py:5159`): + +```python +def select_dispatch_candidates( + ready: list[tuple[Task, str]], +) -> tuple[list[tuple[Task, str]], list[tuple[Task, str]], str]: + ready_reviews = [(task, stage) for task, stage in ready if stage == "review"] + ready_workers = [(task, stage) for task, stage in ready if stage in {"worker", "selfcheck"}] + return ready_reviews + ready_workers, [], "" +``` + +After: + +```python +def select_dispatch_candidates( + store: StateStore, + ready: list[tuple[Task, str]], + *, + persist: bool, +) -> tuple[list[tuple[Task, str]], list[tuple[Task, str, str]], str]: + # Simulate or atomically commit ordered workspace write claims. +``` + +#### Modified Files and Checklist + +- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py`: validate write-set inputs, persist claims, gate admission, adopt active claims, and release verified completions. +- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py`: cover conflict/disjoint admission, persistence, follow-up replacement, task-group scope, and release. + +#### Test Strategy + +Write direct state-store and scheduler tests with temporary repositories. Assert exact canonical-path collision diagnostics, unchanged ownership while a conflicting follow-up waits, persisted claims after reopening the store, cross-task-group collision enforcement in the same workspace, conservative adoption for legacy active tasks, and release only from verified completion paths. Revise the convergence simulation so two tasks with the same target do not overlap while disjoint work still overlaps. + +#### Verification + +Run the full dispatcher unit command in `Final Verification`; all claim lifecycle, scheduler, and convergence assertions must pass. + +### [REFACTOR-3] Align the project dispatcher contract + +#### Problem + +At `SKILL.md:85-89`, the project contract requires parallel dispatch regardless of write set and classifies the modified-file table only as review/stagnation evidence. At `SKILL.md:193-216`, restart and review procedure text repeats that another task's write set must never delay execution. This contradicts the required runtime file-lock policy and leaves workspace ownership of locators unspecified. + +#### Solution + +Document that each valid PLAN claims its canonical targets per physical workspace, collisions wait without becoming predecessor dependencies, claims survive all stages/follow-ups/restarts and release only on verified completion, malformed write sets fail closed, and separate clones/worktrees have independent ledgers. Document that liveness recovery accepts locator/process evidence only for the current workspace identity. + +Before (`SKILL.md:85`): + +```markdown +- Run worker/self-check and official review in parallel when they belong to different dependency-ready tasks, regardless of write-set. +``` + +After: + +```markdown +- Run dependency-ready stages in parallel only after atomically claiming their canonical PLAN write sets in the current physical workspace. +``` + +#### Modified Files and Checklist + +- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md`: replace unsafe concurrency language with workspace claim and locator-ownership contracts. +- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py`: retain the English-policy and deterministic verification checks. + +#### Test Strategy + +Use the existing project-skill narrative test and add precise policy assertions only where needed. Keep artifact prose in English except for exact Korean runtime protocol literals already allowed by the suite. + +#### Verification + +Run the full dispatcher unit command in `Final Verification`; policy-language checks must pass. + +## Modified Files Summary + +| File | Items | +|------|-------| +| `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py` | REFACTOR-1, REFACTOR-2 | +| `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py` | REFACTOR-1, REFACTOR-2, REFACTOR-3 | +| `agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md` | REFACTOR-3 | + +## Final Verification + +Cached output is not applicable to Python `unittest`. + +```bash +python3 -m py_compile agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py +``` + +Expected: exit code `0` with no output. + +```bash +python3 -m unittest discover -s agent-ops/skills/project/orchestrate-agent-task-loop/tests -p 'test_dispatch.py' +``` + +Expected: every dispatcher test passes, no real provider command is invoked, and the final result is `OK`. + +After completing all code changes, fill implementation-owned sections in `CODE_REVIEW-*-G??.md`.