From dde15565a5ea9dc1091ccf7e33b1e06c0bdf5fbf Mon Sep 17 00:00:00 2001 From: toki Date: Sat, 20 Jun 2026 20:47:22 +0900 Subject: [PATCH] refactor: runner proto socket transport hardening - refactor agent components and update tests - Move completed subtasks (02+01_dispatch_state, 03+01,02_runner_actions) to archive - Refactor agent_runner.dart, oto_server_job_client.dart, registration_client.dart - Update smoke tests and server tests for Go services --- .../code_review_cloud_G07_0.log} | 77 ++- .../code_review_local_G06_1.log | 190 +++++++ .../02+01_dispatch_state/complete.log | 51 ++ .../plan_cloud_G07_0.log} | 0 .../02+01_dispatch_state/plan_local_G06_1.log | 109 ++++ .../code_review_cloud_G07_0.log} | 95 +++- .../03+01,02_runner_actions/complete.log | 51 ++ .../plan_cloud_G07_0.log} | 0 apps/runner/lib/oto/agent/agent_runner.dart | 16 +- .../lib/oto/agent/oto_server_job_client.dart | 4 + .../lib/oto/agent/registration_client.dart | 12 + .../oto_server_connection_smoke_test.dart | 162 +++++- .../httpserver/runner_cicd_handlers.go | 4 +- .../core/internal/httpserver/server_test.go | 65 ++- .../core/internal/runnersocket/server_test.go | 496 ++++++++++++++++++ 15 files changed, 1272 insertions(+), 60 deletions(-) rename agent-task/{m-runner-proto-socket-transport-hardening/02+01_dispatch_state/CODE_REVIEW-cloud-G07.md => archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_cloud_G07_0.log} (53%) create mode 100644 agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_local_G06_1.log create mode 100644 agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/complete.log rename agent-task/{m-runner-proto-socket-transport-hardening/02+01_dispatch_state/PLAN-cloud-G07.md => archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_cloud_G07_0.log} (100%) create mode 100644 agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_local_G06_1.log rename agent-task/{m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/CODE_REVIEW-cloud-G07.md => archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/code_review_cloud_G07_0.log} (62%) create mode 100644 agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/complete.log rename agent-task/{m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/PLAN-cloud-G07.md => archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/plan_cloud_G07_0.log} (100%) diff --git a/agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/CODE_REVIEW-cloud-G07.md b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_cloud_G07_0.log similarity index 53% rename from agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/CODE_REVIEW-cloud-G07.md rename to agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_cloud_G07_0.log index 9512a2b..6c11d81 100644 --- a/agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/CODE_REVIEW-cloud-G07.md +++ b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_cloud_G07_0.log @@ -51,41 +51,45 @@ task=m-runner-proto-socket-transport-hardening/02+01_dispatch_state, plan=0, tag | 항목 | 완료 여부 | |------|---------| -| [API-1] socket queued dispatch와 invalid RunRequest failure | [ ] | -| [API-2] socket terminal report와 active execution gate | [ ] | +| [API-1] socket queued dispatch와 invalid RunRequest failure | [x] | +| [API-2] socket terminal report와 active execution gate | [x] | ## 구현 체크리스트 -- [ ] 선행 `01_socket_lifecycle` PASS complete evidence를 확인한다. -- [ ] [API-1] socket queued dispatch와 invalid RunRequest failure 테스트/보강을 구현한다. 검증: `cd services/core && go test -count=1 ./...` -- [ ] [API-2] socket terminal report와 active execution gate 테스트/보강을 구현한다. 검증: `cd services/core && go test -count=1 ./...` -- [ ] runner socket smoke 관련 회귀를 확인한다. 검증: `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. +- [x] 선행 `01_socket_lifecycle` PASS complete evidence를 확인한다. +- [x] [API-1] socket queued dispatch와 invalid RunRequest failure 테스트/보강을 구현한다. 검증: `cd services/core && go test -count=1 ./...` +- [x] [API-2] socket terminal report와 active execution gate 테스트/보강을 구현한다. 검증: `cd services/core && go test -count=1 ./...` +- [x] runner socket smoke 관련 회귀를 확인한다. 검증: `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. ## 코드리뷰 전용 체크리스트 > **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. > 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. -- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. -- [ ] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. -- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G07_N.log`로 아카이브한다. -- [ ] active `PLAN-*-G??.md`를 `plan_cloud_G07_M.log`로 아카이브한다. -- [ ] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G07_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_cloud_G07_M.log`로 아카이브한다. +- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. - [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. - [ ] PASS이면 active task 디렉터리 `agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/`를 `agent-task/archive/YYYY/MM/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. - [ ] PASS이고 task group이 `m-`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. - [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-runner-proto-socket-transport-hardening/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. -- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active plan/review 파일 또는 follow-up plan을 작성하고 `complete.log`를 작성하지 않는다. +- [x] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active plan/review 파일 또는 follow-up plan을 작성하고 `complete.log`를 작성하지 않는다. - [ ] USER_REVIEW이면 `agent-ops/skills/common/code-review/templates/user-review-template.md` 기준으로 `USER_REVIEW.md`를 작성하고 active `PLAN-*.md`, `CODE_REVIEW-*.md`, `complete.log`를 남기지 않는다. ## 계획 대비 변경 사항 -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ +- `server.go`와 `store.go` 수정 없음: 기존 `validTransitions` 테이블과 `handleExecutionReport`의 implicit guard가 이미 올바른 정책을 구현하고 있어 production 코드 변경 없이 테스트만 추가했다. explicit running check 추가는 계획에서 "필요 시"로 명시되었고, 현재 상태에서 테스트가 모두 통과되므로 불필요하다. +- `server_test.go` 한 파일에 API-1과 API-2 모두 추가: 파일 분리 없이 helper 함수(`testServerWithStore`, `registerRunner`, `createQueuedJob`, `setupRunningExecution`)를 공유해 중복을 최소화했다. ## 주요 설계 결정 -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ +- **API-1 dispatch test**: `net.Pipe()`의 remote end를 goroutine으로 drain하여 `client.Send()` 블로킹 없이 `DispatchQueuedJobs()`를 동기 완료시켰다. 상태 전이(queued→running)는 Send() 완료 후 동기적으로 반영된다. +- **API-1 send failure test**: remote end를 Close() 후 dispatch하여 `client.Send()` 오류를 유발했다. `claimNextQueuedJob`은 execution을 running으로 claim한 뒤 send가 실패하므로, 최종 상태는 execution+job 모두 failed임을 검증했다. +- **API-2 guard test**: `handleExecutionReport` 내부 메서드를 직접 호출했다(white-box). socket transport를 통과하는 end-to-end보다 빠르고 결정적이며, 같은 패키지 내 테스트이므로 접근 가능하다. state guard는 `validTransitions` 테이블의 implicit check로 동작하며, queued/succeeded/cancelled 상태 모두 TransitionExecution 실패로 rejection된다. +- **cancelled + terminal report 충돌 정책**: `CancelJobExecution` 후 terminal report를 시도하면 `TransitionExecution(execID, StateSucceeded)`가 "invalid transition from canceled to succeeded" 오류를 반환해 rejection된다. 현재 store 정책을 테스트로 고정했다. ## 사용자 리뷰 요청 @@ -108,26 +112,39 @@ _기본값은 `없음`이다. 구현 중 새 결정이 필요해 보여도 직 ## 검증 결과 -_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ - ### API-1 중간 검증 ```bash $ cd services/core && go test -count=1 ./... -(output) +? github.com/toki/oto/services/core/cmd/oto-core [no test files] +ok github.com/toki/oto/services/core/internal/cicdstate 0.002s +ok github.com/toki/oto/services/core/internal/httpserver 0.082s +ok github.com/toki/oto/services/core/internal/runnerregistry 0.013s +ok github.com/toki/oto/services/core/internal/runnersocket 0.049s +? github.com/toki/oto/services/core/oto [no test files] ``` ### API-2 중간 검증 ```bash $ cd services/core && go test -count=1 ./... -(output) +? github.com/toki/oto/services/core/cmd/oto-core [no test files] +ok github.com/toki/oto/services/core/internal/cicdstate 0.002s +ok github.com/toki/oto/services/core/internal/httpserver 0.082s +ok github.com/toki/oto/services/core/internal/runnerregistry 0.013s +ok github.com/toki/oto/services/core/internal/runnersocket 0.049s +? github.com/toki/oto/services/core/oto [no test files] ``` ### 최종 검증 ```bash $ cd services/core && go test -count=1 ./... -(output) +? github.com/toki/oto/services/core/cmd/oto-core [no test files] +ok github.com/toki/oto/services/core/internal/cicdstate 0.002s +ok github.com/toki/oto/services/core/internal/httpserver 0.082s +ok github.com/toki/oto/services/core/internal/runnerregistry 0.013s +ok github.com/toki/oto/services/core/internal/runnersocket 0.049s +? github.com/toki/oto/services/core/oto [no test files] $ cd apps/runner && dart test test/oto_server_connection_smoke_test.dart -(output) +00:07 +11: All tests passed! ``` --- @@ -135,3 +152,21 @@ $ cd apps/runner && dart test test/oto_server_connection_smoke_test.dart > **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** > If anything is blank, go back and fill it in before saving this file. > Leave review-agent-only sections unchanged. + +## 코드리뷰 결과 + +- 종합 판정: FAIL +- 차원별 평가: + - correctness: Pass + - completeness: Fail + - test coverage: Fail + - API contract: Fail + - code quality: Pass + - plan deviation: Fail + - verification trust: Pass + - spec conformance: Fail +- 발견된 문제: + - Required: [server_test.go](/config/workspace/oto/services/core/internal/runnersocket/server_test.go:238)에서 `TestServerDispatchQueuedJobOverSocket`가 pipe remote end를 단순 drain만 하므로 `DispatchQueuedJobs()`가 실제 `RunRequest`를 socket으로 보냈는지, 그리고 payload의 runner/job/execution id와 run input이 맞는지 검증하지 못한다. 계획은 loopback/fake proto-socket client로 `RunRequest` 수신을 확인하라고 했고, SDD S03도 socket dispatcher가 idle runner에 `RunRequest`를 push한다는 evidence를 요구한다. 이 테스트는 state transition만으로도 통과하므로 socket dispatch 계약 회귀를 놓친다. 수정: 테스트에서 remote side가 정확히 하나의 `otopb.RunRequest`를 수신/파싱하고 runner id, job id, execution id, pipeline yaml/path, variables/command types를 assertion하도록 보강한다. 가능하면 queued job을 2개 두어 한 dispatch pass에서 하나만 claim되고 나머지는 queued로 남는지도 함께 고정한다. + - Required: [server_test.go](/config/workspace/oto/services/core/internal/runnersocket/server_test.go:276)의 실패 케이스는 socket write failure만 검증하고, 계획/SDD가 요구한 invalid `RunRequest`/invalid run input 실패 경로를 검증하지 않는다. `claimNextQueuedJob()`는 invalid `RunInput`일 때 [server.go](/config/workspace/oto/services/core/internal/runnersocket/server.go:299)에서 send 전 오류를 반환하고 [server.go](/config/workspace/oto/services/core/internal/runnersocket/server.go:87) 경로로 job을 failed 처리하므로, 이 정책이 socket dispatch 테스트로 고정되어야 한다. 수정: `PipelineYAML`과 `PipelineYAMLPath`를 동시에 주거나 둘 다 비운 queued job을 만들고 `DispatchQueuedJobs()` 후 job이 failed가 되며 실행 생성 여부가 현재 정책과 일치함을 assertion한다. + - Suggested: [server_test.go](/config/workspace/oto/services/core/internal/runnersocket/server_test.go:311)의 terminal report 테스트는 completed/failed state만 확인하고 계획에서 언급한 report message/log field 보존 증거를 남기지 않는다. 수정: success 또는 failure report에 `Message`를 넣고 `store.GetLogs(execID)`가 해당 line을 포함하는지 assertion한다. +- 다음 단계: FAIL follow-up plan/review를 생성한다. diff --git a/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_local_G06_1.log b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_local_G06_1.log new file mode 100644 index 0000000..6b7bd70 --- /dev/null +++ b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_local_G06_1.log @@ -0,0 +1,190 @@ + + +# Code Review Reference - REVIEW_API + +> **[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 `구현 체크리스트`; 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 by a selected SDD decision or selected Milestone `구현 잠금 > 결정 필요` item, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Environment/secret/service setup, generic scope conflicts, loop exhaustion, and evidence gaps that a follow-up agent can close are normal follow-up issues, not user-review blockers by themselves. +> Do not ask the user directly, present choices in chat, or call `request_user_input` during implementation; record only the linked SDD/Milestone lock decision in `사용자 리뷰 요청` and stop for code-review. +> Finalization (`코드리뷰 결과`, log rename, `complete.log`, archive moves, `코드리뷰 전용 체크리스트`) is review-agent-only, even after compaction/resume. +> Follow the ownership table at the bottom of this file for which sections you own. + +## 개요 + +date=2026-06-20 +task=m-runner-proto-socket-transport-hardening/02+01_dispatch_state, plan=1, tag=REVIEW_API + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/control-plane-product-surface/milestones/runner-proto-socket-transport-hardening.md` +- Task ids: + - `dispatch-state`: queued job dispatch, active execution gate, invalid RunRequest failure, terminal state transition을 socket transport 기준으로 검증한다. +- Completion mode: check-on-pass +- Split dependency: + - `01_socket_lifecycle` PASS complete evidence가 있어야 구현을 시작한다. + +## Spec Targets + +- SDD: `agent-roadmap/sdd/control-plane-product-surface/runner-proto-socket-transport-hardening/SDD.md` +- Acceptance scenarios: + - `S03`: task=`dispatch-state`; evidence=`Socket RunRequest dispatch와 terminal report가 queued/running/completed/failed/cancelled 상태 전이를 검증한다.` +- Completion mode: spec-check-on-pass + +## Archive Evidence Snapshot + +- Current archived plan: `agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_cloud_G07_0.log` +- Current archived review: `agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_cloud_G07_0.log` +- Verdict: FAIL +- Required summary: + - `services/core/internal/runnersocket/server_test.go`의 dispatch success 테스트가 remote pipe를 drain만 해서 실제 `otopb.RunRequest` 수신/필드 검증을 하지 않는다. + - send failure 테스트는 write failure만 검증하고 invalid `RunInput`/invalid `RunRequest` 실패 정책을 socket dispatch 경로에서 고정하지 않는다. +- Suggested summary: + - terminal report 테스트가 report message/log field 보존 증거를 남기지 않는다. +- Affected files: + - `services/core/internal/runnersocket/server_test.go` + - 필요 시 `services/core/internal/runnersocket/server.go` +- Verification evidence from failed loop: + - `cd services/core && go test -count=1 ./...` PASS + - `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` PASS + - `git diff --check` PASS +- Roadmap/spec carryover: + - Roadmap task `dispatch-state`와 SDD S03 completion target은 유지한다. + - 선행 `01_socket_lifecycle` PASS evidence는 `agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/01_socket_lifecycle/complete.log`에서 확인되었다. +- Allowed narrow archive reread: + - 위 archived plan/review log와 선행 dependency `complete.log`만 필요 시 좁게 읽는다. `agent-task/archive/**` 전체 검색은 하지 않는다. + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. +리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다. + +1. 판정을 append한다. +2. `CODE_REVIEW-local-G06.md` -> `code_review_local_G06_N.log`, `PLAN-local-G06.md` -> `plan_local_G06_M.log`로 아카이브한다. +3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다. +4. PASS이고 task group이 `m-`이면 완료 이벤트 메타데이터를 보고한다. roadmap 상태 체크와 `update-roadmap` 호출은 런타임 책임이다. +5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REVIEW_API-1] socket dispatch `RunRequest` payload와 single-claim 증거 | [x] | +| [REVIEW_API-2] invalid run input failed 정책과 terminal report message/log 증거 | [x] | + +## 구현 체크리스트 + +- [x] [REVIEW_API-1] socket dispatch `RunRequest` payload와 single-claim 증거를 보강한다. 검증: `cd services/core && go test -count=1 ./internal/runnersocket` +- [x] [REVIEW_API-2] invalid run input failed 정책과 terminal report message/log 증거를 보강한다. 검증: `cd services/core && go test -count=1 ./internal/runnersocket` +- [x] 전체 core/runner smoke 회귀를 확인한다. 검증: `cd services/core && go test -count=1 ./...` 및 `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. +> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. + +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_local_G06_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_local_G06_M.log`로 아카이브한다. +- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. +- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [x] PASS이면 active task 디렉터리 `agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/`를 `agent-task/archive/YYYY/MM/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [x] PASS이고 task group이 `m-`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-runner-proto-socket-transport-hardening/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active plan/review 파일 또는 follow-up plan을 작성하고 `complete.log`를 작성하지 않는다. +- [ ] USER_REVIEW이면 `agent-ops/skills/common/code-review/templates/user-review-template.md` 기준으로 `USER_REVIEW.md`를 작성하고 active `PLAN-*.md`, `CODE_REVIEW-*.md`, `complete.log`를 남기지 않는다. + +## 계획 대비 변경 사항 + +- 계획: `TestServerDispatchQueuedJobOverSocket`를 보강한다. 실제: 기존 테스트를 인라인 job 생성 + `pipeClientPair` 사용으로 전면 교체해 payload 검증 포함. `createQueuedJob` 헬퍼 대신 Variables/CommandTypes를 포함한 `store.CreateJob`을 직접 호출. 별도 테스트 함수를 추가하지 않고 기존 함수를 강화함으로써 중복 제거. +- 계획: "기존 `proto_socket` helper로 파싱하기 어렵다면 package-local helper를 작게 만든다" → `AddListenerTyped[*otopb.RunRequest]`로 충분히 파싱 가능하므로 별도 helper 없이 `pipeClientPair` + `runnerParserMap`으로 해결. + +## 주요 설계 결정 + +- `runnerParserMap()`: runner 수신 방향에 필요한 `RunRequest`만 포함. `CancelRunRequest` 등 미사용 타입은 추가하지 않음. +- `pipeClientPair()`: `net.Pipe()` 양 끝을 각각 `TcpClient`로 래핑. `intervalSec=0`으로 heartbeat 비활성화해 타이머 노이즈 없음. server-side는 기존 `parserMap()`, runner-side는 `runnerParserMap()` 사용. +- invalid run input: `validateRunInput` 실패 시 `execID=""`이므로 execution이 생성되지 않는다는 기존 정책을 두 fixture(both-empty, both-set)로 고정. production 코드 변경 없이 테스트만 추가. +- terminal report message: `handleExecutionReport`가 `Message != ""` 이면 `store.AppendLog`를 호출하는 기존 동작을 success/failure 두 케이스로 고정. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 새 결정이 필요해 보여도 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 이 섹션은 선택된 SDD 결정 또는 선택된 Milestone `구현 잠금 > 결정 필요` 항목이 실구현을 차단할 때만 채운다. 외부 환경/secret/서비스 준비, 검증 증거 공백, 반복 실패, 일반 범위 조정은 사용자 리뷰 요청이 아니며 `검증 결과`, `계획 대비 변경 사항`, 또는 code-review의 일반 follow-up plan으로 처리한다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 연결 대상: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- dispatch success 테스트가 실제 socket `RunRequest`를 수신/파싱하고 payload를 검증하는지 확인한다. +- 한 runner의 한 dispatch pass가 queued job 하나만 claim하고 나머지 queued job을 남기는지 확인한다. +- invalid run input 실패 정책이 socket dispatch 경로에서 테스트로 고정되었는지 확인한다. +- terminal report message가 execution log로 보존되는지 확인한다. + +## 검증 결과 + +_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ + +필수 규칙: +- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다. +- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다. +- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. +- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다. + +### REVIEW_API-1 중간 검증 +```bash +$ cd services/core && go test -count=1 ./internal/runnersocket +ok github.com/toki/oto/services/core/internal/runnersocket 0.046s +``` + +### REVIEW_API-2 중간 검증 +```bash +$ cd services/core && go test -count=1 ./internal/runnersocket +ok github.com/toki/oto/services/core/internal/runnersocket 0.046s +``` + +### 최종 검증 +```bash +$ cd services/core && go test -count=1 ./... +? github.com/toki/oto/services/core/cmd/oto-core [no test files] +ok github.com/toki/oto/services/core/internal/cicdstate 0.002s +ok github.com/toki/oto/services/core/internal/httpserver 0.066s +ok github.com/toki/oto/services/core/internal/runnerregistry 0.006s +ok github.com/toki/oto/services/core/internal/runnersocket 0.048s +? github.com/toki/oto/services/core/oto [no test files] +$ cd apps/runner && dart test test/oto_server_connection_smoke_test.dart +00:08 +11: All tests passed! +``` + +--- + +> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** +> If anything is blank, go back and fill it in before saving this file. +> Leave review-agent-only sections unchanged. + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가: + - correctness: Pass + - completeness: Pass + - test coverage: Pass + - API contract: Pass + - code quality: Pass + - plan deviation: Pass + - verification trust: Pass + - spec conformance: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS complete.log 작성 후 task directory를 archive로 이동한다. diff --git a/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/complete.log b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/complete.log new file mode 100644 index 0000000..ddb93f5 --- /dev/null +++ b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/complete.log @@ -0,0 +1,51 @@ +# Complete - m-runner-proto-socket-transport-hardening/02+01_dispatch_state + +## 완료 일시 + +2026-06-20 + +## 요약 + +Runner proto-socket dispatch/state 보강을 2회 리뷰 루프로 완료했다. 최종 판정은 PASS다. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_cloud_G07_0.log` | `code_review_cloud_G07_0.log` | FAIL | socket `RunRequest` payload 수신 검증, invalid input 실패 정책, terminal report message log 증거가 부족해 follow-up 필요 | +| `plan_local_G06_1.log` | `code_review_local_G06_1.log` | PASS | socket payload, single-claim, invalid input failure, terminal report message log 보존 테스트를 보강 | + +## 구현/정리 내용 + +- `services/core/internal/runnersocket/server_test.go`에 runner-side proto-socket parser를 둔 `pipeClientPair` helper를 추가해 실제 `RunRequest` payload를 수신/검증했다. +- dispatch success, one-job-per-runner claim, send failure, invalid run input failure, terminal report state guard, terminal report message log 보존 테스트를 socket dispatch/state evidence로 정리했다. +- production code 변경 없이 기존 `DispatchQueuedJobs`, `claimNextQueuedJob`, `handleExecutionReport` 정책을 테스트로 고정했다. + +## 최종 검증 + +- `cd services/core && go test -count=1 ./internal/runnersocket` - PASS; `ok github.com/toki/oto/services/core/internal/runnersocket 0.049s` +- `cd services/core && go test -count=1 ./...` - PASS; core packages all passed, including `internal/runnersocket` +- `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` - PASS; `00:12 +11: All tests passed!` +- `git diff --check` - PASS; no whitespace errors + +## Roadmap Completion + +- Milestone: `agent-roadmap/phase/control-plane-product-surface/milestones/runner-proto-socket-transport-hardening.md` +- Completed task ids: + - `dispatch-state`: PASS; evidence=`agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_local_G06_1.log`, `agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_local_G06_1.log`; verification=`cd services/core && go test -count=1 ./internal/runnersocket`, `cd services/core && go test -count=1 ./...`, `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`, `git diff --check` +- Not completed task ids: 없음 + +## Spec Completion + +- SDD: `agent-roadmap/sdd/control-plane-product-surface/runner-proto-socket-transport-hardening/SDD.md` +- Completed scenario ids: + - `S03`: PASS; task=`dispatch-state`; evidence=`agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_local_G06_1.log`, `agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_local_G06_1.log`; verification=`cd services/core && go test -count=1 ./internal/runnersocket`, `cd services/core && go test -count=1 ./...`, `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`, `git diff --check` +- Not completed scenario ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/PLAN-cloud-G07.md b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_cloud_G07_0.log similarity index 100% rename from agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/PLAN-cloud-G07.md rename to agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_cloud_G07_0.log diff --git a/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_local_G06_1.log b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_local_G06_1.log new file mode 100644 index 0000000..e8a7f5c --- /dev/null +++ b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_local_G06_1.log @@ -0,0 +1,109 @@ + + +# Plan - REVIEW_API + +## 이 파일을 읽는 구현 에이전트에게 + +`CODE_REVIEW-local-G06.md`의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채우는 것까지가 구현입니다. 검증을 실행하고 실제 stdout/stderr를 붙인 뒤 active 파일을 그대로 두고 review ready로 보고하세요. 선택된 SDD 결정 또는 Milestone `구현 잠금 > 결정 필요` 항목이 막으면 review stub의 `사용자 리뷰 요청` 섹션을 채우고 멈추며, 직접 사용자에게 질문하거나 `USER_REVIEW.md`, `complete.log`, archive를 만들지 않습니다. 환경/secret/서비스 문제나 증거 공백은 일반 follow-up 대상으로 기록합니다. + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/control-plane-product-surface/milestones/runner-proto-socket-transport-hardening.md` +- Task ids: + - `dispatch-state`: queued job dispatch, active execution gate, invalid RunRequest failure, terminal state transition을 socket transport 기준으로 검증한다. +- Completion mode: check-on-pass +- Split dependency: + - `01_socket_lifecycle` PASS complete evidence가 있어야 구현을 시작한다. + +## Spec Targets + +- SDD: `agent-roadmap/sdd/control-plane-product-surface/runner-proto-socket-transport-hardening/SDD.md` +- Acceptance scenarios: + - `S03`: task=`dispatch-state`; evidence=`Socket RunRequest dispatch와 terminal report가 queued/running/completed/failed/cancelled 상태 전이를 검증한다.` +- Completion mode: spec-check-on-pass + +## Archive Evidence Snapshot + +- Current archived plan: `agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_cloud_G07_0.log` +- Current archived review: `agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_cloud_G07_0.log` +- Verdict: FAIL +- Required summary: + - `services/core/internal/runnersocket/server_test.go`의 dispatch success 테스트가 remote pipe를 drain만 해서 실제 `otopb.RunRequest` 수신/필드 검증을 하지 않는다. + - send failure 테스트는 write failure만 검증하고 invalid `RunInput`/invalid `RunRequest` 실패 정책을 socket dispatch 경로에서 고정하지 않는다. +- Suggested summary: + - terminal report 테스트가 report message/log field 보존 증거를 남기지 않는다. +- Affected files: + - `services/core/internal/runnersocket/server_test.go` + - 필요 시 `services/core/internal/runnersocket/server.go` +- Verification evidence from failed loop: + - `cd services/core && go test -count=1 ./...` PASS + - `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` PASS + - `git diff --check` PASS +- Roadmap/spec carryover: + - Roadmap task `dispatch-state`와 SDD S03 completion target은 유지한다. + - 선행 `01_socket_lifecycle` PASS evidence는 `agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/01_socket_lifecycle/complete.log`에서 확인되었다. +- Allowed narrow archive reread: + - 위 archived plan/review log와 선행 dependency `complete.log`만 필요 시 좁게 읽는다. `agent-task/archive/**` 전체 검색은 하지 않는다. + +## 배경 + +이 follow-up은 production behavior 변경이 아니라 SDD S03과 `dispatch-state` 완료 근거를 충족하기 위한 테스트 증거 보강이다. 기존 구현은 state transition test를 추가했지만, socket payload 자체와 invalid input 실패 정책을 충분히 관찰하지 못했다. + +## 범위 결정 근거 + +- 포함: `services/core/internal/runnersocket/server_test.go`의 Go test 보강. +- 필요 시 포함: 테스트 가능성을 위해 아주 작은 helper 추가 또는 `server.go`의 오류 정책 명시 보강. +- 제외: cancel/status/self-update operator action, HTTP compatibility boundary, Dart runner implementation 변경. +- 사용자 리뷰 요청 금지: 이번 범위는 SDD/Milestone 미정 결정이 아니라 테스트 증거 공백이므로 파일 기반 follow-up으로 해결한다. + +## 구현 체크리스트 + +- [ ] [REVIEW_API-1] socket dispatch `RunRequest` payload와 single-claim 증거를 보강한다. 검증: `cd services/core && go test -count=1 ./internal/runnersocket` +- [ ] [REVIEW_API-2] invalid run input failed 정책과 terminal report message/log 증거를 보강한다. 검증: `cd services/core && go test -count=1 ./internal/runnersocket` +- [ ] 전체 core/runner smoke 회귀를 확인한다. 검증: `cd services/core && go test -count=1 ./...` 및 `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +### [REVIEW_API-1] socket dispatch `RunRequest` payload와 single-claim 증거 + +문제: `TestServerDispatchQueuedJobOverSocket`는 `net.Pipe()`의 remote side를 drain만 하므로 `DispatchQueuedJobs()`가 실제 runner-facing `RunRequest`를 push했는지 검증하지 않는다. state transition만 확인하면 socket payload regression을 놓칠 수 있다. + +해결 방법: + +- remote side에서 proto-socket frame을 실제로 읽어 `otopb.RunRequest`를 수신했음을 검증한다. 기존 `proto_socket` helper로 파싱하기 어렵다면 package-local helper를 작게 만들되, 테스트 전용으로 제한한다. +- 수신한 `RunRequest`의 `RunnerId`, `JobId`, `ExecutionId`, `PipelineYaml` 또는 `PipelineYamlPath`, `Variables`, `CommandTypes`를 store state와 대조한다. +- queued job을 2개 준비하고 한 runner에 대해 `DispatchQueuedJobs()` 한 번이 하나만 claim하는지 확인한다. 첫 job은 running, 두 번째 job은 queued로 남아야 한다. + +중간 검증: + +```bash +cd services/core && go test -count=1 ./internal/runnersocket +``` + +기대 결과: `internal/runnersocket` package 통과. + +### [REVIEW_API-2] invalid run input failed 정책과 terminal report message/log 증거 + +문제: 기존 실패 테스트는 socket write failure만 다루고, SDD S03의 invalid input -> failed 증거를 socket dispatch 경로에서 남기지 않는다. terminal report 테스트도 state만 확인하고 report message가 log로 보존되는지 확인하지 않는다. + +해결 방법: + +- invalid run input fixture를 만든다. 예: `PipelineYAML`과 `PipelineYAMLPath`를 동시에 설정하거나 둘 다 비운 queued job. +- `DispatchQueuedJobs()` 후 현재 정책을 명확히 고정한다. 현 production 코드가 유지된다면 invalid input은 send 전 실패하므로 job은 failed가 되고 execution은 생성되지 않는 것이 기대값이다. 정책을 바꾸는 경우에는 execution failed까지 코드와 테스트를 함께 맞춘다. +- terminal report success 또는 failure case에 `Message`를 넣고 `store.GetLogs(execID)`가 해당 message를 포함하는지 검증한다. + +중간 검증: + +```bash +cd services/core && go test -count=1 ./internal/runnersocket +``` + +기대 결과: `internal/runnersocket` package 통과. + +## 최종 검증 + +```bash +cd services/core && go test -count=1 ./... +cd apps/runner && dart test test/oto_server_connection_smoke_test.dart +``` + +기대 결과: Go tests 통과, runner socket smoke 통과. 모든 코드 변경 완료 후 반드시 `CODE_REVIEW-local-G06.md`의 구현 에이전트 소유 섹션을 채운다. diff --git a/agent-task/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/CODE_REVIEW-cloud-G07.md b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/code_review_cloud_G07_0.log similarity index 62% rename from agent-task/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/CODE_REVIEW-cloud-G07.md rename to agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/code_review_cloud_G07_0.log index b49b48d..6d86381 100644 --- a/agent-task/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/CODE_REVIEW-cloud-G07.md +++ b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/code_review_cloud_G07_0.log @@ -52,42 +52,47 @@ task=m-runner-proto-socket-transport-hardening/03+01,02_runner_actions, plan=0, | 항목 | 완료 여부 | |------|---------| -| [API-1] cancel/status action의 socket-first 방향과 상태 guard | [ ] | -| [API-2] runner-side cancel handling 또는 compatibility 제외 | [ ] | -| [API-3] self-update action 방향 명확화 | [ ] | +| [API-1] cancel/status action의 socket-first 방향과 상태 guard | [x] | +| [API-2] runner-side cancel handling 또는 compatibility 제외 | [x] | +| [API-3] self-update action 방향 명확화 | [x] | ## 구현 체크리스트 -- [ ] 선행 `01_socket_lifecycle`, `02+01_dispatch_state` PASS complete evidence를 확인한다. -- [ ] [API-1] cancel/status action의 socket-first 방향과 상태 guard를 구현/검증한다. 검증: `cd services/core && go test -count=1 ./...` -- [ ] [API-2] runner-side cancel handling 또는 명시적 compatibility 제외를 구현/검증한다. 검증: `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` -- [ ] [API-3] self-update action을 socket-first 구현 또는 compatibility 제외 범위로 명확히 한다. 검증: `cd apps/runner && dart analyze` -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. +- [x] 선행 `01_socket_lifecycle`, `02+01_dispatch_state` PASS complete evidence를 확인한다. +- [x] [API-1] cancel/status action의 socket-first 방향과 상태 guard를 구현/검증한다. 검증: `cd services/core && go test -count=1 ./...` +- [x] [API-2] runner-side cancel handling 또는 명시적 compatibility 제외를 구현/검증한다. 검증: `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` +- [x] [API-3] self-update action을 socket-first 구현 또는 compatibility 제외 범위로 명확히 한다. 검증: `cd apps/runner && dart analyze` +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. ## 코드리뷰 전용 체크리스트 > **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. > 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. -- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. -- [ ] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. -- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G07_N.log`로 아카이브한다. -- [ ] active `PLAN-*-G??.md`를 `plan_cloud_G07_M.log`로 아카이브한다. -- [ ] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. -- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. -- [ ] PASS이면 active task 디렉터리 `agent-task/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/`를 `agent-task/archive/YYYY/MM/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. -- [ ] PASS이고 task group이 `m-`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. -- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-runner-proto-socket-transport-hardening/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G07_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_cloud_G07_M.log`로 아카이브한다. +- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. +- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [x] PASS이면 active task 디렉터리 `agent-task/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/`를 `agent-task/archive/YYYY/MM/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [x] PASS이고 task group이 `m-`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-runner-proto-socket-transport-hardening/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. - [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active plan/review 파일 또는 follow-up plan을 작성하고 `complete.log`를 작성하지 않는다. - [ ] USER_REVIEW이면 `agent-ops/skills/common/code-review/templates/user-review-template.md` 기준으로 `USER_REVIEW.md`를 작성하고 active `PLAN-*.md`, `CODE_REVIEW-*.md`, `complete.log`를 남기지 않는다. ## 계획 대비 변경 사항 -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ +- `CancelRunRequest` runner-side 수신은 실제 프로세스 중단까지 확장하지 않고 socket stream 수신과 log 처리로 제한했다. 현재 `RemoteRunExecutor`/`Application.build()`에는 cooperative cancel hook이 없어서 실행 중단은 후속 범위로 남기고, 이번 작업은 server store cancel과 late terminal report guard를 source of truth로 둔다. +- socket cancel smoke는 처음 HTTP claim으로 running execution을 만들려 했으나, socket runner 연결 상태에서는 job 생성 직후 server가 자동 dispatch하는 것이 정상 동작이라 `runRequests` stream에서 server-assigned execution id를 받아 cancel하는 방식으로 수정했다. +- self-update는 socket command로 구현하지 않고 HTTP compatibility action으로 명시했다. 응답 메시지와 Dart client 주석에서 idle accept/active defer만 이번 범위임을 드러냈다. ## 주요 설계 결정 -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ +- operator cancel HTTP action은 store를 `canceled` terminal state로 전환한 뒤 runner socket에 best-effort `CancelRunRequest`를 보낸다. socket send 실패는 cancel 응답 성공 여부를 뒤집지 않으며, terminal conflict는 store transition guard가 막는다. +- status는 runner socket command가 아니라 registry/store read query로 유지한다. 테스트에서 status path가 dispatcher를 호출하지 않음을 고정했다. +- Dart push session은 `runRequests`와 별도 `cancelRequests` stream을 노출한다. 기본 agent loop는 cancel 수신을 log로 남기고, terminal conflict 처리는 server state guard에 맡긴다. +- self-update는 `SelfUpdateRequest` proto가 남아 있어도 이번 Milestone에서는 socket dispatch 대상이 아니다. HTTP compatibility response가 idle accepted 또는 active deferred 결과만 반환한다. ## 사용자 리뷰 요청 @@ -115,29 +120,54 @@ _구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 ### API-1 중간 검증 ```bash $ cd services/core && go test -count=1 ./... -(output) +? github.com/toki/oto/services/core/cmd/oto-core [no test files] +ok github.com/toki/oto/services/core/internal/cicdstate 0.002s +ok github.com/toki/oto/services/core/internal/httpserver 0.082s +ok github.com/toki/oto/services/core/internal/runnerregistry 0.003s +ok github.com/toki/oto/services/core/internal/runnersocket 0.048s +? github.com/toki/oto/services/core/oto [no test files] ``` ### API-2 중간 검증 ```bash $ cd apps/runner && dart test test/oto_server_connection_smoke_test.dart -(output) +00:00 +0: loading test/oto_server_connection_smoke_test.dart +00:00 +0: OTO Dart runner registers with Go OTO Server, goes online, and disconnects on close +... +00:10 +12: OtoServerSocketRegistrationClient socket session close is idempotent after server disconnect +00:10 +12: All tests passed! ``` ### API-3 중간 검증 ```bash $ cd apps/runner && dart analyze -(output) +Analyzing runner... +No issues found! ``` ### 최종 검증 ```bash $ cd services/core && go test -count=1 ./... -(output) +? github.com/toki/oto/services/core/cmd/oto-core [no test files] +ok github.com/toki/oto/services/core/internal/cicdstate 0.002s +ok github.com/toki/oto/services/core/internal/httpserver 0.082s +ok github.com/toki/oto/services/core/internal/runnerregistry 0.003s +ok github.com/toki/oto/services/core/internal/runnersocket 0.048s +? github.com/toki/oto/services/core/oto [no test files] + $ cd apps/runner && dart test test/oto_server_connection_smoke_test.dart -(output) +00:00 +0: loading test/oto_server_connection_smoke_test.dart +00:00 +0: OTO Dart runner registers with Go OTO Server, goes online, and disconnects on close +... +00:10 +12: OtoServerSocketRegistrationClient socket session close is idempotent after server disconnect +00:10 +12: All tests passed! + $ cd apps/runner && dart analyze -(output) +Analyzing runner... +No issues found! + +$ git diff --check +(no output; exit 0) ``` --- @@ -145,3 +175,18 @@ $ cd apps/runner && dart analyze > **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** > If anything is blank, go back and fill it in before saving this file. > Leave review-agent-only sections unchanged. + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가: + - correctness: Pass + - completeness: Pass + - test coverage: Pass + - API contract: Pass + - code quality: Pass + - plan deviation: Pass + - verification trust: Pass + - spec conformance: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS이므로 active plan/review 파일을 로그로 아카이브하고 `complete.log`를 작성한 뒤 task 디렉터리를 archive로 이동한다. diff --git a/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/complete.log b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/complete.log new file mode 100644 index 0000000..7ac0594 --- /dev/null +++ b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/complete.log @@ -0,0 +1,51 @@ +# Complete - m-runner-proto-socket-transport-hardening/03+01,02_runner_actions + +## 완료 일시 + +2026-06-20 + +## 요약 + +Runner action 방향성 정리 작업을 1회 리뷰 루프로 완료했다. 최종 판정은 PASS다. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_cloud_G07_0.log` | `code_review_cloud_G07_0.log` | PASS | cancel/status/self-update의 socket-first 및 HTTP compatibility 경계가 코드, 테스트, 검증 결과로 충족됨 | + +## 구현/정리 내용 + +- operator cancel HTTP action이 store를 terminal canceled로 전환한 뒤 runner socket으로 best-effort `CancelRunRequest`를 전달하도록 검증했다. +- cancel 이후 late terminal report가 canceled 상태를 뒤집지 못하도록 state guard를 테스트로 고정했다. +- status는 socket command가 아닌 registry/store read query로 유지하고, self-update는 HTTP compatibility action으로 명시했다. +- Dart socket push session에 `cancelRequests` stream을 추가해 socket cancel 수신을 smoke로 검증했다. + +## 최종 검증 + +- `cd services/core && go test -count=1 ./...` - PASS; core packages all passed. +- `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` - PASS; `+12: All tests passed!` +- `cd apps/runner && dart analyze` - PASS; `No issues found!` +- `git diff --check` - PASS; no output. + +## Roadmap Completion + +- Milestone: `agent-roadmap/phase/control-plane-product-surface/milestones/runner-proto-socket-transport-hardening.md` +- Completed task ids: + - `runner-actions`: PASS; evidence=`agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/plan_cloud_G07_0.log`, `agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/code_review_cloud_G07_0.log`; verification=`cd services/core && go test -count=1 ./...`, `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`, `cd apps/runner && dart analyze`, `git diff --check` +- Not completed task ids: 없음 + +## Spec Completion + +- SDD: `agent-roadmap/sdd/control-plane-product-surface/runner-proto-socket-transport-hardening/SDD.md` +- Completed scenario ids: + - `S04`: PASS; task=`runner-actions`; evidence=`agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/plan_cloud_G07_0.log`, `agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/code_review_cloud_G07_0.log`; verification=`cd services/core && go test -count=1 ./...`, `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`, `cd apps/runner && dart analyze`, `git diff --check` +- Not completed scenario ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/PLAN-cloud-G07.md b/agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/plan_cloud_G07_0.log similarity index 100% rename from agent-task/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/PLAN-cloud-G07.md rename to agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/plan_cloud_G07_0.log diff --git a/apps/runner/lib/oto/agent/agent_runner.dart b/apps/runner/lib/oto/agent/agent_runner.dart index 33b60fc..532db6d 100644 --- a/apps/runner/lib/oto/agent/agent_runner.dart +++ b/apps/runner/lib/oto/agent/agent_runner.dart @@ -106,7 +106,7 @@ class DefaultAgentRunner implements AgentRunner { ); var runChain = Future.value(); - final subscription = session.runRequests.listen((request) { + final runSubscription = session.runRequests.listen((request) { runChain = runChain.then((_) async { if (request.jobId.isEmpty || request.executionId.isEmpty) { _log('Received RunRequest without job_id or execution_id; skipping.'); @@ -125,10 +125,22 @@ class DefaultAgentRunner implements AgentRunner { } }); }); + final cancelSubscription = session.cancelRequests.listen((request) { + final executionId = request.executionId; + if (executionId.isEmpty) { + _log('Received CancelRunRequest without execution_id; skipping.'); + return; + } + _log( + 'Received cancel for execution $executionId; ' + 'server state guard owns terminal conflict handling.', + ); + }); _log('Agent connected via proto-socket. Waiting for pushed jobs...'); await shutdownFuture; - await subscription.cancel(); + await runSubscription.cancel(); + await cancelSubscription.cancel(); await runChain; } diff --git a/apps/runner/lib/oto/agent/oto_server_job_client.dart b/apps/runner/lib/oto/agent/oto_server_job_client.dart index 40efc8d..e0b6196 100644 --- a/apps/runner/lib/oto/agent/oto_server_job_client.dart +++ b/apps/runner/lib/oto/agent/oto_server_job_client.dart @@ -343,6 +343,10 @@ class OtoServerJobClient implements RemoteJobReporter { return RunnerStatusResult.fromJson(response); } + /// Requests the operator-facing HTTP compatibility self-update action. + /// + /// This milestone does not dispatch a socket self-update command; the server + /// either accepts an idle compatibility request or defers while a run is active. Future requestSelfUpdate({ required String version, required String downloadUrl, diff --git a/apps/runner/lib/oto/agent/registration_client.dart b/apps/runner/lib/oto/agent/registration_client.dart index bd4c42d..52103c6 100644 --- a/apps/runner/lib/oto/agent/registration_client.dart +++ b/apps/runner/lib/oto/agent/registration_client.dart @@ -103,6 +103,7 @@ abstract class OtoServerJobSession implements AgentSession { abstract class OtoServerPushJobSession implements AgentSession { RemoteJobReporter get jobReporter; Stream get runRequests; + Stream get cancelRequests; } abstract class RegistrationClient { @@ -319,6 +320,8 @@ class _OtoServerSocketAgentSession implements OtoServerPushJobSession { final String _runnerId; final StreamController _runRequests = StreamController(); + final StreamController _cancelRequests = + StreamController(); Timer? _heartbeatTimer; bool _closed = false; @@ -331,6 +334,9 @@ class _OtoServerSocketAgentSession implements OtoServerPushJobSession { @override Stream get runRequests => _runRequests.stream; + @override + Stream get cancelRequests => _cancelRequests.stream; + _OtoServerSocketAgentSession( this.result, this._client, @@ -346,6 +352,11 @@ class _OtoServerSocketAgentSession implements OtoServerPushJobSession { _runRequests.add(request); } }); + _client.addListener((request) { + if (!_cancelRequests.isClosed) { + _cancelRequests.add(request); + } + }); if (result.accepted) { _heartbeatTimer = Timer.periodic( heartbeatInterval, @@ -376,6 +387,7 @@ class _OtoServerSocketAgentSession implements OtoServerPushJobSession { _heartbeatTimer?.cancel(); _heartbeatTimer = null; unawaited(_runRequests.close()); + unawaited(_cancelRequests.close()); await _client.close(); } } diff --git a/apps/runner/test/oto_server_connection_smoke_test.dart b/apps/runner/test/oto_server_connection_smoke_test.dart index 97d328e..99b25e6 100644 --- a/apps/runner/test/oto_server_connection_smoke_test.dart +++ b/apps/runner/test/oto_server_connection_smoke_test.dart @@ -5,6 +5,7 @@ import 'dart:io'; import 'package:http/http.dart' as http; import 'package:test/test.dart'; import 'package:oto/oto/agent/agent_config.dart'; +import 'package:oto/oto/agent/oto_server_job_client.dart'; import 'package:oto/oto/agent/registration_client.dart'; import 'package:oto/oto/agent/remote_run_executor.dart'; import 'package:oto/oto/application.dart'; @@ -1398,11 +1399,17 @@ pipeline: expect(session.result.accepted, isTrue); expect(session.result.runnerId, _runnerId); - // Capture runRequests stream completion after close(). - final streamDone = Completer(); - final sub = (session as OtoServerPushJobSession).runRequests.listen( + // Capture stream completion after close(). + final pushSession = session as OtoServerPushJobSession; + final runRequestsDone = Completer(); + final cancelRequestsDone = Completer(); + final runSub = pushSession.runRequests.listen( (_) {}, - onDone: streamDone.complete, + onDone: runRequestsDone.complete, + ); + final cancelSub = pushSession.cancelRequests.listen( + (_) {}, + onDone: cancelRequestsDone.complete, ); final httpClient = http.Client(); @@ -1419,13 +1426,21 @@ pipeline: reason: 'socket runner did not go online in registry', ); - // close() must cancel the heartbeat timer and close runRequests. + // close() must cancel the heartbeat timer and close socket streams. await session.close(); - await streamDone.future.timeout( + await runRequestsDone.future.timeout( const Duration(seconds: 2), - onTimeout: () => fail('runRequests stream did not close after session.close()'), + onTimeout: () => + fail('runRequests stream did not close after session.close()'), ); - await sub.cancel(); + await cancelRequestsDone.future.timeout( + const Duration(seconds: 2), + onTimeout: () => fail( + 'cancelRequests stream did not close after session.close()', + ), + ); + await runSub.cancel(); + await cancelSub.cancel(); final isDisconnected = await _pollRunnerStatus( httpClient, @@ -1458,6 +1473,137 @@ pipeline: timeout: const Timeout(Duration(seconds: 30)), ); + test( + 'OtoServerSocketRegistrationClient receives socket cancel requests from HTTP cancel action', + () async { + final port = await _freePort(); + final serverAddr = '$_host:$port'; + const jobId = 'smoke-job-socket-cancel'; + + final (:process, :socketAddr) = await _startCoreWithSocket(serverAddr); + + final output = StringBuffer(); + final stdoutSub = process.stdout + .transform(systemEncoding.decoder) + .listen(output.write, onError: output.write); + final stderrSub = process.stderr + .transform(systemEncoding.decoder) + .listen(output.write, onError: output.write); + + try { + await _waitForPort(_host, port, process, output); + + final agentConfig = AgentConfig( + agent: const AgentIdentityConfig( + id: _runnerId, + alias: _runnerAlias, + enrollmentToken: _token, + ), + server: ServerConnectionConfig( + url: 'http://$serverAddr', + socketUrl: 'tcp://$socketAddr', + ), + runtime: const AgentRuntimeConfig( + installDir: '/tmp/install', + workspaceRoot: '/tmp/workspace', + logDir: '/tmp/log', + ), + ); + + final client = OtoServerSocketRegistrationClient( + commandTypes: ['Shell'], + heartbeatInterval: const Duration(milliseconds: 200), + ); + final session = await client.openSession(agentConfig); + expect(session.result.accepted, isTrue); + final pushSession = session as OtoServerPushJobSession; + + final runReceived = Completer(); + final runSub = pushSession.runRequests.listen((request) { + if (!runReceived.isCompleted) { + expect(request.runnerId, _runnerId); + expect(request.jobId, jobId); + expect(request.executionId, isNotEmpty); + runReceived.complete(request.executionId); + } + }); + var expectedExecutionId = ''; + final cancelReceived = Completer(); + final cancelSub = pushSession.cancelRequests.listen((request) { + if (!cancelReceived.isCompleted) { + expect(request.runnerId, _runnerId); + expect(request.executionId, expectedExecutionId); + expect(request.reason, 'socket cancel smoke'); + cancelReceived.complete(request.executionId); + } + }); + + final httpClient = http.Client(); + final jobs = OtoServerJobClient( + serverUrl: 'http://$serverAddr', + runnerId: _runnerId, + client: httpClient, + ); + try { + final createResp = await httpClient.post( + Uri.parse('http://$serverAddr/api/v1/jobs'), + headers: {'content-type': 'application/json'}, + body: jsonEncode({ + 'id': jobId, + 'name': 'socket cancel smoke', + 'run_request': { + 'pipeline_yaml': 'commands:\n - type: Shell', + 'command_types': ['Shell'], + }, + }), + ); + expect(createResp.statusCode, equals(201)); + + expectedExecutionId = await runReceived.future.timeout( + const Duration(seconds: 3), + onTimeout: () => fail('socket RunRequest was not received'), + ); + + final cancelRes = await jobs.cancelRun( + executionId: expectedExecutionId, + reason: 'socket cancel smoke', + ); + expect(cancelRes.success, isTrue); + + await cancelReceived.future.timeout( + const Duration(seconds: 3), + onTimeout: () => fail('socket CancelRunRequest was not received'), + ); + + final execResp = await httpClient.get( + Uri.parse( + 'http://$serverAddr/api/v1/executions/$expectedExecutionId', + ), + ); + final execJson = jsonDecode(execResp.body) as Map; + expect(execJson['state'], equals('canceled')); + } finally { + await runSub.cancel(); + await cancelSub.cancel(); + httpClient.close(); + await session.close(); + } + } finally { + process.kill(ProcessSignal.sigterm); + await process.exitCode.timeout( + const Duration(seconds: 5), + onTimeout: () { + process.kill(ProcessSignal.sigkill); + return process.exitCode; + }, + ); + await stdoutSub.cancel(); + await stderrSub.cancel(); + } + }, + timeout: const Timeout(Duration(seconds: 30)), + ); + test( 'OtoServerSocketRegistrationClient socket session close is idempotent after server disconnect', () async { diff --git a/services/core/internal/httpserver/runner_cicd_handlers.go b/services/core/internal/httpserver/runner_cicd_handlers.go index 7219cae..c5acd6a 100644 --- a/services/core/internal/httpserver/runner_cicd_handlers.go +++ b/services/core/internal/httpserver/runner_cicd_handlers.go @@ -474,10 +474,10 @@ func handleRunnerSelfUpdate(store *cicdstate.Store, registry *runnerregistry.Reg } if hasActiveExecution { - writeResponse(w, http.StatusOK, selfUpdateToJSON(false, false, true, req.GetVersion(), "deferred: runner is currently running a job")) + writeResponse(w, http.StatusOK, selfUpdateToJSON(false, false, true, req.GetVersion(), "deferred: runner is currently running a job; self-update remains an HTTP compatibility action")) return } - writeResponse(w, http.StatusOK, selfUpdateToJSON(true, true, false, req.GetVersion(), "self-update request accepted")) + writeResponse(w, http.StatusOK, selfUpdateToJSON(true, true, false, req.GetVersion(), "self-update compatibility request accepted; no socket self-update command was dispatched")) } } diff --git a/services/core/internal/httpserver/server_test.go b/services/core/internal/httpserver/server_test.go index 31d1408..12c9c92 100644 --- a/services/core/internal/httpserver/server_test.go +++ b/services/core/internal/httpserver/server_test.go @@ -18,6 +18,27 @@ import ( otopb "github.com/toki/oto/services/core/oto" ) +type runnerCancelCall struct { + runnerID string + execID string + reason string +} + +type recordingRunnerControlDispatcher struct { + dispatchCount int + cancelCalls []runnerCancelCall + cancelErr error +} + +func (d *recordingRunnerControlDispatcher) DispatchQueuedJobs() { + d.dispatchCount++ +} + +func (d *recordingRunnerControlDispatcher) CancelExecution(runnerID, execID, reason string) error { + d.cancelCalls = append(d.cancelCalls, runnerCancelCall{runnerID: runnerID, execID: execID, reason: reason}) + return d.cancelErr +} + func TestHandleHealthz(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "/healthz", nil) rr := httptest.NewRecorder() @@ -1489,6 +1510,7 @@ func TestHandleRunnerRegisterRejectsIncompatibleCapability(t *testing.T) { func TestHandleRunnerCancelExecution(t *testing.T) { store := cicdstate.NewStore() registry := runnerregistry.New() + dispatcher := &recordingRunnerControlDispatcher{} registry.Register(&otopb.RegisterRunnerRequest{ EnrollmentToken: "token-123", @@ -1507,7 +1529,7 @@ func TestHandleRunnerCancelExecution(t *testing.T) { body := bytes.NewBufferString(`{"runner_id":"runner-1", "execution_id":"exec-1", "reason":"user request"}`) req := httptest.NewRequest(http.MethodPost, "/api/v1/runners/runner-1/executions/exec-1/cancel", body) rr := httptest.NewRecorder() - handleRouter(store, registry)(rr, req) + handleRouterWithDispatcher(store, registry, dispatcher)(rr, req) if rr.Code != http.StatusOK { t.Fatalf("cancel status = %v, want %v; body=%s", rr.Code, http.StatusOK, rr.Body.String()) @@ -1517,11 +1539,40 @@ func TestHandleRunnerCancelExecution(t *testing.T) { if !cancelResp.Success { t.Fatal("expected success cancel response") } + if len(dispatcher.cancelCalls) != 1 { + t.Fatalf("cancel dispatcher calls = %d, want 1", len(dispatcher.cancelCalls)) + } + call := dispatcher.cancelCalls[0] + if call.runnerID != "runner-1" || call.execID != "exec-1" || call.reason != "user request" { + t.Fatalf("cancel dispatcher call = %+v, want runner-1/exec-1/user request", call) + } job, _ := store.GetJob("job-1") if job.State != cicdstate.StateCanceled { t.Fatalf("job state = %q, want canceled", job.State) } + exec, _ := store.GetExecution("exec-1") + if exec.State != cicdstate.StateCanceled { + t.Fatalf("execution state = %q, want canceled", exec.State) + } + + // Late terminal reports from the runner must not overwrite the canceled + // terminal state established by the operator action. + lateReportBody := bytes.NewBufferString(`{"runner_id":"runner-1", "execution_id":"exec-1", "job_id":"job-1", "success":true}`) + lateReportReq := httptest.NewRequest(http.MethodPost, "/api/v1/runners/runner-1/executions/exec-1/report", lateReportBody) + lateReportRR := httptest.NewRecorder() + handleRouterWithDispatcher(store, registry, dispatcher)(lateReportRR, lateReportReq) + if lateReportRR.Code != http.StatusBadRequest { + t.Fatalf("late report status = %v, want %v; body=%s", lateReportRR.Code, http.StatusBadRequest, lateReportRR.Body.String()) + } + job, _ = store.GetJob("job-1") + if job.State != cicdstate.StateCanceled { + t.Fatalf("late report changed job state = %q, want canceled", job.State) + } + exec, _ = store.GetExecution("exec-1") + if exec.State != cicdstate.StateCanceled { + t.Fatalf("late report changed execution state = %q, want canceled", exec.State) + } // 2. Mismatch cancellation bodyMismatch := bytes.NewBufferString(`{"runner_id":"runner-wrong", "execution_id":"exec-1"}`) @@ -1585,6 +1636,7 @@ func TestHandleRunnerCancelExecution(t *testing.T) { func TestHandleRunnerStatus(t *testing.T) { store := cicdstate.NewStore() registry := runnerregistry.New() + dispatcher := &recordingRunnerControlDispatcher{} registry.Register(&otopb.RegisterRunnerRequest{ EnrollmentToken: "token-123", @@ -1599,7 +1651,7 @@ func TestHandleRunnerStatus(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "/api/v1/runners/runner-1/status", nil) rr := httptest.NewRecorder() - handleRouter(store, registry)(rr, req) + handleRouterWithDispatcher(store, registry, dispatcher)(rr, req) if rr.Code != http.StatusOK { t.Fatalf("status code = %v, want %v", rr.Code, http.StatusOK) @@ -1616,6 +1668,9 @@ func TestHandleRunnerStatus(t *testing.T) { if res["current_job_id"] != "job-1" { t.Fatalf("current_job_id = %v", res["current_job_id"]) } + if dispatcher.dispatchCount != 0 || len(dispatcher.cancelCalls) != 0 { + t.Fatalf("status query must not dispatch socket commands; dispatcher=%+v", dispatcher) + } } func TestHandleRunnerSelfUpdate(t *testing.T) { @@ -1651,6 +1706,9 @@ func TestHandleRunnerSelfUpdate(t *testing.T) { if resOk["accepted"] != true || resOk["deferred"] != false { t.Fatalf("idle update response = %+v", resOk) } + if !strings.Contains(fmt.Sprint(resOk["message"]), "compatibility") { + t.Fatalf("idle update message must mark compatibility scope, got %+v", resOk) + } // 3. Active execution defer store.CreateJob("job-1", "build", nil) @@ -1671,6 +1729,9 @@ func TestHandleRunnerSelfUpdate(t *testing.T) { if resActive["accepted"] != false || resActive["deferred"] != true { t.Fatalf("active update response = %+v", resActive) } + if !strings.Contains(fmt.Sprint(resActive["message"]), "HTTP compatibility") { + t.Fatalf("active update message must mark HTTP compatibility scope, got %+v", resActive) + } } func waitServerReady(t *testing.T, baseURL string) { diff --git a/services/core/internal/runnersocket/server_test.go b/services/core/internal/runnersocket/server_test.go index e664eee..fccd38d 100644 --- a/services/core/internal/runnersocket/server_test.go +++ b/services/core/internal/runnersocket/server_test.go @@ -26,6 +26,29 @@ func pipeClient(t *testing.T) (*protoSocket.TcpClient, net.Conn) { return client, b } +// runnerParserMap returns a parser map for the runner (client) side of a +// proto-socket connection. It includes server-to-runner frames so tests can +// verify dispatch and cancel delivery. +func runnerParserMap() protoSocket.ParserMap { + return protoSocket.ParserMap{ + protoSocket.TypeNameOf(&otopb.RunRequest{}): parserFor(func() *otopb.RunRequest { return &otopb.RunRequest{} }), + protoSocket.TypeNameOf(&otopb.CancelRunRequest{}): parserFor(func() *otopb.CancelRunRequest { return &otopb.CancelRunRequest{} }), + } +} + +// pipeClientPair creates two TcpClients backed by the two ends of a net.Pipe. +// serverSide uses parserMap(); runnerSide uses runnerParserMap() so tests can +// capture and inspect RunRequest frames sent by DispatchQueuedJobs. +// Cleanup closes both connections. +func pipeClientPair(t *testing.T) (serverSide, runnerSide *protoSocket.TcpClient) { + t.Helper() + a, b := net.Pipe() + serverSide = protoSocket.NewTcpClient(a, 0, 0, parserMap()) + runnerSide = protoSocket.NewTcpClient(b, 0, 0, runnerParserMap()) + t.Cleanup(func() { _ = a.Close(); _ = b.Close() }) + return serverSide, runnerSide +} + // TestServerSetClientReplacesDuplicate verifies that registering a second client // for the same runner ID closes the old one and stores the new one. func TestServerReplacesDuplicateRunnerClient(t *testing.T) { @@ -176,6 +199,479 @@ func TestServerDuplicateReplaceWithListenerDoesNotDeadlock(t *testing.T) { } } +// testServerWithStore creates a Server sharing the given registry and returns +// the associated Store for state inspection in tests. +func testServerWithStore(registry *runnerregistry.Registry) (*Server, *cicdstate.Store) { + store := cicdstate.NewStore() + return New("127.0.0.1:0", registry, store), store +} + +// registerRunner adds a runner to the registry with default test fields. +func registerRunner(t *testing.T, registry *runnerregistry.Registry, runnerID string) { + t.Helper() + resp := registry.Register(&otopb.RegisterRunnerRequest{ + EnrollmentToken: "tok", + RunnerId: runnerID, + ProtocolVersion: "oto.runner.v1", + Capability: &otopb.RunnerCapability{Name: "oto-runner", Version: "1.0.0"}, + }) + if !resp.GetAccepted() { + t.Fatalf("registerRunner %s: rejected: %s", runnerID, resp.GetRejectReason()) + } +} + +// createQueuedJob adds a job with a valid inline YAML to the store. +func createQueuedJob(t *testing.T, store *cicdstate.Store, jobID string) { + t.Helper() + if _, err := store.CreateJob(jobID, jobID, &cicdstate.RunInput{PipelineYAML: "pipeline: test"}); err != nil { + t.Fatalf("CreateJob %s: %v", jobID, err) + } +} + +// setupRunningExecution creates a queued job, registers the runner, creates an +// execution, and transitions both job and execution to running. Returns execID. +func setupRunningExecution(t *testing.T, store *cicdstate.Store, registry *runnerregistry.Registry, jobID, runnerID string) string { + t.Helper() + createQueuedJob(t, store, jobID) + registerRunner(t, registry, runnerID) + execID := store.NextExecutionID() + if _, err := store.CreateExecution(jobID, execID); err != nil { + t.Fatalf("CreateExecution: %v", err) + } + _ = store.SetExecutionRunnerID(execID, runnerID) + if err := store.TransitionJob(jobID, cicdstate.StateRunning); err != nil { + t.Fatalf("TransitionJob running: %v", err) + } + if err := store.TransitionExecution(execID, cicdstate.StateRunning); err != nil { + t.Fatalf("TransitionExecution running: %v", err) + } + return execID +} + +// TestServerDispatchQueuedJobOverSocket verifies that DispatchQueuedJobs +// transitions the queued job and its execution to running over the socket +// transport, assigns the dispatched runner ID to the execution, and sends a +// RunRequest frame whose fields match the store state. +func TestServerDispatchQueuedJobOverSocket(t *testing.T) { + registry := runnerregistry.New() + s, store := testServerWithStore(registry) + registerRunner(t, registry, "runner-dispatch") + + jobID := "job-dispatch" + if _, err := store.CreateJob(jobID, jobID, &cicdstate.RunInput{ + PipelineYAML: "pipeline: dispatch-test", + Variables: map[string]string{"ENV": "ci"}, + CommandTypes: []string{"flutter"}, + }); err != nil { + t.Fatalf("CreateJob: %v", err) + } + + // Use a runner-side TcpClient so we can parse the RunRequest frame. + serverClient, runnerClient := pipeClientPair(t) + + received := make(chan *otopb.RunRequest, 1) + protoSocket.AddListenerTyped[*otopb.RunRequest](&runnerClient.Communicator, func(req *otopb.RunRequest) { + select { + case received <- req: + default: + } + }) + + s.setClient("runner-dispatch", serverClient) + s.DispatchQueuedJobs() + + var rr *otopb.RunRequest + select { + case rr = <-received: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for RunRequest") + } + + job, err := store.GetJob(jobID) + if err != nil { + t.Fatalf("GetJob: %v", err) + } + if job.State != cicdstate.StateRunning { + t.Errorf("job state: want %s, got %s", cicdstate.StateRunning, job.State) + } + if job.ExecutionID == "" { + t.Fatal("no execution created for dispatched job") + } + exec, err := store.GetExecution(job.ExecutionID) + if err != nil { + t.Fatalf("GetExecution: %v", err) + } + if exec.State != cicdstate.StateRunning { + t.Errorf("execution state: want %s, got %s", cicdstate.StateRunning, exec.State) + } + if exec.RunnerID != "runner-dispatch" { + t.Errorf("execution runnerID: want runner-dispatch, got %s", exec.RunnerID) + } + + // Verify RunRequest payload fields match the store state. + if rr.RunnerId != "runner-dispatch" { + t.Errorf("RunRequest.RunnerId: want runner-dispatch, got %s", rr.RunnerId) + } + if rr.JobId != jobID { + t.Errorf("RunRequest.JobId: want %s, got %s", jobID, rr.JobId) + } + if rr.ExecutionId != job.ExecutionID { + t.Errorf("RunRequest.ExecutionId: want %s, got %s", job.ExecutionID, rr.ExecutionId) + } + if rr.PipelineYaml != "pipeline: dispatch-test" { + t.Errorf("RunRequest.PipelineYaml: want %q, got %q", "pipeline: dispatch-test", rr.PipelineYaml) + } + if len(rr.Variables) == 0 || rr.Variables["ENV"] != "ci" { + t.Errorf("RunRequest.Variables: want {ENV:ci}, got %v", rr.Variables) + } + if len(rr.CommandTypes) != 1 || rr.CommandTypes[0] != "flutter" { + t.Errorf("RunRequest.CommandTypes: want [flutter], got %v", rr.CommandTypes) + } +} + +// TestServerCancelExecutionSendsSocketCancelToRunner verifies the server-side +// cancel action uses the runner-facing socket as a best-effort signal channel. +func TestServerCancelExecutionSendsSocketCancelToRunner(t *testing.T) { + s := testServer(runnerregistry.New()) + serverClient, runnerClient := pipeClientPair(t) + + received := make(chan *otopb.CancelRunRequest, 1) + protoSocket.AddListenerTyped[*otopb.CancelRunRequest](&runnerClient.Communicator, func(req *otopb.CancelRunRequest) { + select { + case received <- req: + default: + } + }) + + s.setClient("runner-cancel", serverClient) + + if err := s.CancelExecution("runner-cancel", "exec-cancel", "operator request"); err != nil { + t.Fatalf("CancelExecution: %v", err) + } + + var req *otopb.CancelRunRequest + select { + case req = <-received: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for CancelRunRequest") + } + if req.RunnerId != "runner-cancel" { + t.Errorf("CancelRunRequest.RunnerId: want runner-cancel, got %s", req.RunnerId) + } + if req.ExecutionId != "exec-cancel" { + t.Errorf("CancelRunRequest.ExecutionId: want exec-cancel, got %s", req.ExecutionId) + } + if req.Reason != "operator request" { + t.Errorf("CancelRunRequest.Reason: want operator request, got %s", req.Reason) + } + + if err := s.CancelExecution("runner-missing", "exec-cancel", "operator request"); err == nil { + t.Fatal("expected missing connected runner to return an error") + } +} + +// TestServerDispatchClaimsOnlyOneJobPerRunner verifies that a single +// DispatchQueuedJobs call dispatches at most one job per runner, leaving any +// remaining queued jobs untouched. +func TestServerDispatchClaimsOnlyOneJobPerRunner(t *testing.T) { + registry := runnerregistry.New() + s, store := testServerWithStore(registry) + registerRunner(t, registry, "runner-claim") + createQueuedJob(t, store, "job-claim-1") + createQueuedJob(t, store, "job-claim-2") + + client, remote := pipeClient(t) + // Drain the remote end so Send does not block. + go func() { + buf := make([]byte, 4096) + for { + if _, err := remote.Read(buf); err != nil { + return + } + } + }() + + s.setClient("runner-claim", client) + s.DispatchQueuedJobs() + + job1, err := store.GetJob("job-claim-1") + if err != nil { + t.Fatalf("GetJob job-claim-1: %v", err) + } + job2, err := store.GetJob("job-claim-2") + if err != nil { + t.Fatalf("GetJob job-claim-2: %v", err) + } + + running, queued := 0, 0 + for _, j := range []*cicdstate.Job{job1, job2} { + switch j.State { + case cicdstate.StateRunning: + running++ + case cicdstate.StateQueued: + queued++ + } + } + if running != 1 { + t.Errorf("want exactly 1 running job after one dispatch pass, got %d", running) + } + if queued != 1 { + t.Errorf("want exactly 1 queued job after one dispatch pass, got %d", queued) + } +} + +// TestServerFailsExecutionWhenRunRequestCannotBeSent verifies that a send +// error over the socket transitions the claimed execution and job to failed +// rather than leaving them stuck in running. +func TestServerFailsExecutionWhenRunRequestCannotBeSent(t *testing.T) { + registry := runnerregistry.New() + s, store := testServerWithStore(registry) + registerRunner(t, registry, "runner-sendfail") + createQueuedJob(t, store, "job-sendfail") + + client, remote := pipeClient(t) + // Close the remote end before dispatch so any write to the client fails. + _ = remote.Close() + + s.setClient("runner-sendfail", client) + s.DispatchQueuedJobs() + + job, err := store.GetJob("job-sendfail") + if err != nil { + t.Fatalf("GetJob: %v", err) + } + if job.State != cicdstate.StateFailed { + t.Errorf("job state: want %s, got %s", cicdstate.StateFailed, job.State) + } + if job.ExecutionID == "" { + t.Fatal("expected execution to be created before send failure") + } + exec, err := store.GetExecution(job.ExecutionID) + if err != nil { + t.Fatalf("GetExecution: %v", err) + } + if exec.State != cicdstate.StateFailed { + t.Errorf("execution state: want %s, got %s", cicdstate.StateFailed, exec.State) + } +} + +// TestServerReportsRunResultOverSocket verifies that handleExecutionReport +// accepts success and failure terminal reports from the owning runner and +// transitions execution and job to the correct terminal state. +func TestServerReportsRunResultOverSocket(t *testing.T) { + cases := []struct { + name string + success bool + wantState string + }{ + {"success", true, cicdstate.StateSucceeded}, + {"failure", false, cicdstate.StateFailed}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + registry := runnerregistry.New() + s, store := testServerWithStore(registry) + jobID := "job-report-" + tc.name + runnerID := "runner-report-" + tc.name + execID := setupRunningExecution(t, store, registry, jobID, runnerID) + + resp := s.handleExecutionReport(&otopb.ExecutionReportRequest{ + RunnerId: runnerID, + JobId: jobID, + ExecutionId: execID, + Success: tc.success, + }) + + if !resp.GetAccepted() { + t.Fatalf("report rejected: %s", resp.GetErrorMessage()) + } + exec, err := store.GetExecution(execID) + if err != nil { + t.Fatalf("GetExecution: %v", err) + } + if exec.State != tc.wantState { + t.Errorf("execution state: want %s, got %s", tc.wantState, exec.State) + } + job, err := store.GetJob(jobID) + if err != nil { + t.Fatalf("GetJob: %v", err) + } + if job.State != tc.wantState { + t.Errorf("job state: want %s, got %s", tc.wantState, job.State) + } + }) + } +} + +// TestServerRejectsTerminalReportForInactiveExecution verifies the active +// execution gate: terminal reports on queued, succeeded, or cancelled +// executions are rejected, and reports from the wrong runner are rejected. +func TestServerRejectsTerminalReportForInactiveExecution(t *testing.T) { + t.Run("queued_execution", func(t *testing.T) { + registry := runnerregistry.New() + s, store := testServerWithStore(registry) + registerRunner(t, registry, "runner-guard") + createQueuedJob(t, store, "job-guard-queued") + execID := store.NextExecutionID() + if _, err := store.CreateExecution("job-guard-queued", execID); err != nil { + t.Fatalf("CreateExecution: %v", err) + } + _ = store.SetExecutionRunnerID(execID, "runner-guard") + // Execution remains in queued state; terminal report must be rejected. + resp := s.handleExecutionReport(&otopb.ExecutionReportRequest{ + RunnerId: "runner-guard", JobId: "job-guard-queued", ExecutionId: execID, Success: true, + }) + if resp.GetAccepted() { + t.Error("expected rejection for queued execution, got accepted") + } + }) + + t.Run("succeeded_execution", func(t *testing.T) { + registry := runnerregistry.New() + s, store := testServerWithStore(registry) + execID := setupRunningExecution(t, store, registry, "job-guard-succ", "runner-guard-succ") + _ = store.TransitionExecution(execID, cicdstate.StateSucceeded) + _ = store.TransitionJob("job-guard-succ", cicdstate.StateSucceeded) + + resp := s.handleExecutionReport(&otopb.ExecutionReportRequest{ + RunnerId: "runner-guard-succ", JobId: "job-guard-succ", ExecutionId: execID, Success: true, + }) + if resp.GetAccepted() { + t.Error("expected rejection for succeeded execution, got accepted") + } + }) + + t.Run("cancelled_execution", func(t *testing.T) { + registry := runnerregistry.New() + s, store := testServerWithStore(registry) + execID := setupRunningExecution(t, store, registry, "job-guard-cancel", "runner-guard-cancel") + if err := store.CancelJobExecution("job-guard-cancel", execID); err != nil { + t.Fatalf("CancelJobExecution: %v", err) + } + resp := s.handleExecutionReport(&otopb.ExecutionReportRequest{ + RunnerId: "runner-guard-cancel", JobId: "job-guard-cancel", ExecutionId: execID, Success: true, + }) + if resp.GetAccepted() { + t.Error("expected rejection for cancelled execution, got accepted") + } + }) + + t.Run("wrong_runner", func(t *testing.T) { + registry := runnerregistry.New() + s, store := testServerWithStore(registry) + execID := setupRunningExecution(t, store, registry, "job-guard-runner", "runner-owner") + registerRunner(t, registry, "runner-other") + + resp := s.handleExecutionReport(&otopb.ExecutionReportRequest{ + RunnerId: "runner-other", JobId: "job-guard-runner", ExecutionId: execID, Success: true, + }) + if resp.GetAccepted() { + t.Error("expected rejection for wrong runner, got accepted") + } + if resp.GetErrorMessage() == "" { + t.Error("expected non-empty error message for wrong runner") + } + }) +} + +// TestServerDispatchInvalidRunInputFails verifies that DispatchQueuedJobs +// transitions a job with an invalid RunInput directly to failed without creating +// an execution. Two invalid fixture variants are covered: both fields empty and +// both fields set simultaneously. +func TestServerDispatchInvalidRunInputFails(t *testing.T) { + cases := []struct { + name string + input *cicdstate.RunInput + }{ + { + "both_empty", + &cicdstate.RunInput{PipelineYAML: "", PipelineYAMLPath: ""}, + }, + { + "both_set", + &cicdstate.RunInput{PipelineYAML: "pipeline: test", PipelineYAMLPath: "/some/path.yaml"}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + registry := runnerregistry.New() + s, store := testServerWithStore(registry) + runnerID := "runner-invalid-" + tc.name + jobID := "job-invalid-" + tc.name + registerRunner(t, registry, runnerID) + + if _, err := store.CreateJob(jobID, jobID, tc.input); err != nil { + t.Fatalf("CreateJob: %v", err) + } + + // No drain needed: validateRunInput fails before client.Send is called. + client, _ := pipeClient(t) + s.setClient(runnerID, client) + s.DispatchQueuedJobs() + + job, err := store.GetJob(jobID) + if err != nil { + t.Fatalf("GetJob: %v", err) + } + if job.State != cicdstate.StateFailed { + t.Errorf("job state: want %s, got %s", cicdstate.StateFailed, job.State) + } + if job.ExecutionID != "" { + t.Errorf("no execution should be created for invalid run input; got execID %s", job.ExecutionID) + } + }) + } +} + +// TestServerReportMessageSavedToLog verifies that a non-empty Message in an +// ExecutionReportRequest is appended to the execution log after the report is +// accepted. +func TestServerReportMessageSavedToLog(t *testing.T) { + cases := []struct { + name string + success bool + message string + }{ + {"success_with_message", true, "build succeeded: artifacts uploaded"}, + {"failure_with_message", false, "build failed: compilation error"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + registry := runnerregistry.New() + s, store := testServerWithStore(registry) + jobID := "job-log-" + tc.name + runnerID := "runner-log-" + tc.name + execID := setupRunningExecution(t, store, registry, jobID, runnerID) + + resp := s.handleExecutionReport(&otopb.ExecutionReportRequest{ + RunnerId: runnerID, + JobId: jobID, + ExecutionId: execID, + Success: tc.success, + Message: tc.message, + }) + if !resp.GetAccepted() { + t.Fatalf("report rejected: %s", resp.GetErrorMessage()) + } + + logs, err := store.GetLogs(execID) + if err != nil { + t.Fatalf("GetLogs: %v", err) + } + found := false + for _, entry := range logs { + if entry.Line == tc.message { + found = true + break + } + } + if !found { + t.Errorf("expected execution log to contain %q; got entries: %v", tc.message, logs) + } + }) + } +} + // TestServerHeartbeatUsesRegisteredRunnerID verifies that connectedRunnerIDs // reflects only currently active clients and that clientFor returns nil after // the current client is removed.