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

chore: update control-plane edge wire baseline and related components

Browse source 
- Move completed review/plan artifacts to archive
- Update control-plane main, wire, edge components
- Update edge bootstrap runtime and controlplane connector
- Update generated proto files
This commit is contained in:
toki 2026-05-30 22:30:18 +09:00
parent ef8d0c4f7a
commit 7df7c6f52e
23 changed files with 1749 additions and 250 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_0.log → agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_0.log
View file

95
agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/CODE_REVIEW-cloud-G08.md → agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_1.log
View file

@ -41,39 +41,42 @@ task=m-control-plane-edge-wire-baseline/04_status_snapshot, plan=1, tag=REVIEW_S
| 항목 | 완료 여부 |
|------|---------|
| [REVIEW_STATUS-1] Proto Generated Artifact Idempotence | [ ] |
| [REVIEW_STATUS-1] Proto Generated Artifact Idempotence | [x] |
## 구현 체크리스트
- [ ] 현재 repo 툴체인으로 `make proto`를 실행해 `proto/gen/iop/control.pb.go`가 idempotent한 generated output이 되도록 맞춘다.
- [ ] repo 밖 temp dir에 Makefile과 동일한 protoc 명령을 실행해 `proto/gen/iop/control.pb.go`와 temp generated file의 diff가 없는지 확인하고, `node.pb.go`/`runtime.pb.go`/`job.pb.go`에 불필요한 diff가 남지 않았음을 기록한다.
- [ ] focused Go test와 Control Plane-Edge wire smoke를 재실행하고 실제 stdout/stderr를 review stub에 기록한다.
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
- [x] 현재 repo 툴체인으로 `make proto`를 실행해 `proto/gen/iop/control.pb.go`가 idempotent한 generated output이 되도록 맞춘다.
- [x] repo 밖 temp dir에 Makefile과 동일한 protoc 명령을 실행해 `proto/gen/iop/control.pb.go`와 temp generated file의 diff가 없는지 확인하고, `node.pb.go`/`runtime.pb.go`/`job.pb.go`에 불필요한 diff가 남지 않았음을 기록한다.
- [x] focused Go test와 Control Plane-Edge wire smoke를 재실행하고 실제 stdout/stderr를 review stub에 기록한다.
- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
## 코드리뷰 전용 체크리스트
> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다.
> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다.
- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
- [ ] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다.
- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G08_N.log`로 아카이브한다.
- [ ] active `PLAN-*-G??.md`를 `plan_cloud_G08_M.log`로 아카이브한다.
- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다.
- [ ] PASS이면 active task 디렉터리 `agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/`를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/04_status_snapshot/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다.
- [ ] PASS이고 task group이 `m-control-plane-edge-wire-baseline`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-control-plane-edge-wire-baseline/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다.
- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다.
- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G08_N.log`로 아카이브한다.
- [x] active `PLAN-*-G??.md`를 `plan_cloud_G08_M.log`로 아카이브한다.
- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다.
- [x] PASS이면 active task 디렉터리 `agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/`를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/04_status_snapshot/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다.
- [x] PASS이고 task group이 `m-control-plane-edge-wire-baseline`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-control-plane-edge-wire-baseline/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다.
- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성하고 `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`를 남기지 않는다.
- [ ] USER_REVIEW가 사용자 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다.
## 계획 대비 변경 사항
_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._
- 이전 plan=0에서 `make proto`를 PATH 끝에 `/config/.local/bin`을 붙여 실행한 탓에 `/usr/bin/protoc`(libprotoc 3.21.12)가 선택되어 `control.pb.go` 헤더가 `protoc v3.21.12`로 다운그레이드됐고, 그 상태로 커밋(`348578e`)되어 idempotence FAIL이 났다. 이번 follow-up에서 원인을 정확히 식별했다: 이 환경에는 protoc 두 개가 있다 — `/usr/bin/protoc` = libprotoc 3.21.12, `/config/.local/bin/protoc` = libprotoc 29.3. repo의 다른 생성물(`node/runtime/job.pb.go`) 헤더가 모두 `protoc v5.29.3`이므로 repo-canonical 툴체인은 libprotoc 29.3이다.
- 따라서 검증은 repo-canonical 29.3 protoc가 선택되도록 `PATH="/config/.local/bin:$PATH"`를 prepend해 실행했다. 명령 자체(`protoc --version`, `make proto`, temp diff, `git diff --name-only`, go test, `make test-control-plane-edge-wire`)는 계획의 고정 계약 그대로이며, 대체/생략은 없다. PATH prepend는 명령 변경이 아니라 의도한 툴체인 선택이다. (FAIL 리뷰가 본 `protoc --version` = libprotoc 29.3과 동일한 바이너리.)
- 코드/계약 변경은 없다. `control.pb.go`의 유일한 diff는 헤더의 `protoc v3.21.12` → `protoc v5.29.3` 1줄이며, 메시지 본문(EdgeStatusRequest/EdgeNodeSnapshot/EdgeStatusResponse)은 protoc-gen-go v1.36.11이 동일하게 생성한다. `node.pb.go`/`runtime.pb.go`/`job.pb.go`에는 churn이 없다.
## 주요 설계 결정
_구현 에이전트가 주요 설계 결정 사항을 기록한다._
- generated artifact는 repo-canonical 툴체인(libprotoc 29.3 + protoc-gen-go v1.36.11)으로만 갱신해 헤더 버전을 다른 생성물과 일치시키고 `make proto` idempotence를 보장한다. 더 오래된 `/usr/bin/protoc`(3.21.12)로 생성하지 않는다.
- 이전 리뷰의 의미 구현(STATUS-1~4: proto 계약, Control Plane request path, Edge status provider)은 `348578e`에 그대로 유지하고, 이번 follow-up은 generated header/idempotence와 검증 기록 신뢰도만 회복한다. status 계약, wire, connector 로직은 의미 변경하지 않는다.
## 사용자 리뷰 요청
@ -104,33 +107,67 @@ _구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후
- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다.
- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다.
> 주: 이 환경에는 protoc 두 개가 있고 repo-canonical 툴체인은 libprotoc 29.3(`/config/.local/bin/protoc`)이다. 아래 명령은 모두 `PATH="/config/.local/bin:$PATH"`로 canonical protoc가 선택된 셸에서 실행했다(계획 대비 변경 사항 참조).
### REVIEW_STATUS-1 중간 검증
```text
$ protoc --version
(output)
libprotoc 29.3
$ protoc-gen-go --version
(output)
protoc-gen-go v1.36.11
$ make proto
(output)
protoc \
--go_out=. \
--go_opt=module=iop \
--proto_path=. \
proto/iop/runtime.proto \
proto/iop/node.proto \
proto/iop/control.proto \
proto/iop/job.proto
(exit 0)
$ tmpdir=$(mktemp -d); protoc --go_out="$tmpdir" --go_opt=module=iop --proto_path=. proto/iop/runtime.proto proto/iop/node.proto proto/iop/control.proto proto/iop/job.proto && diff -u proto/gen/iop/control.pb.go "$tmpdir/proto/gen/iop/control.pb.go"; status=$?; rm -rf "$tmpdir"; exit $status
(output)
DIFF_EXIT=0
(diff 출력 없음 = control.pb.go가 현재 toolchain 생성물과 idempotent)
$ git diff --name-only -- proto/gen/iop/node.pb.go proto/gen/iop/runtime.pb.go proto/gen/iop/job.pb.go
(output)
(출력 없음 = status 계약과 무관한 generated churn 없음)
```
`control.pb.go`의 working-tree diff(`348578e` 대비)는 헤더 1줄뿐이다:
```text
$ git diff proto/gen/iop/control.pb.go | head -12
@@ -1,7 +1,7 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.11
-// protoc v3.21.12
+// protoc v5.29.3
// source: proto/iop/control.proto
```
### 최종 검증
```text
$ go test -count=1 ./apps/control-plane/internal/wire ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./proto/gen/...
(output)
ok iop/apps/control-plane/internal/wire 0.748s
ok iop/apps/edge/internal/controlplane 4.415s
ok iop/apps/edge/internal/bootstrap 0.014s
? iop/proto/gen/iop [no test files]
$ make test-control-plane-edge-wire
(output)
[cp-edge-wire] CP: hello accepted
[cp-edge-wire] Edge: connected to control plane
[cp-edge-wire] CP: edge disconnected
{"msg":"edge hello accepted","edge_id":"smoke-edge-wire","edge_name":"Smoke Edge Wire","version":"0.1.0"}
{"msg":"edge disconnected","edge_id":"smoke-edge-wire","reason":"remote_closed"}
{"logger":"controlplane","msg":"connected to control plane","wire_addr":"127.0.0.1:31875","protocol":"protobuf-socket"}
[cp-edge-wire] Control Plane-Edge wire smoke PASSED.
(exit 0)
```
---
@ -153,3 +190,17 @@ $ make test-control-plane-edge-wire
| 리뷰어를 위한 체크포인트 | 스텁 생성 시 고정 | 리뷰 초점 |
| 검증 결과 | 구현 에이전트 | 실제 stdout/stderr 기록 |
| 코드리뷰 결과 | 리뷰 에이전트 | 스텁에 포함하지 않음 |
## 코드리뷰 결과
- 종합 판정: PASS
- 차원별 평가:
- correctness: Pass
- completeness: Pass
- test coverage: Pass
- API contract: Pass
- code quality: Pass
- plan deviation: Pass
- verification trust: Pass
- 발견된 문제: 없음
- 다음 단계: PASS이므로 `complete.log` 작성 후 active task 디렉터리를 archive로 이동한다.

46
agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/complete.log Normal file
View file

@ -0,0 +1,46 @@
# Complete - m-control-plane-edge-wire-baseline/04_status_snapshot
## 완료 일시
2026-05-30
## 요약
Control Plane-Edge status snapshot wire baseline follow-up completed after 2 review loops; final verdict PASS.
## 루프 이력
| Plan | Review | Verdict | 메모 |
|------|--------|---------|------|
| `plan_cloud_G08_0.log` | `code_review_cloud_G08_0.log` | FAIL | Status snapshot implementation was functionally sound, but `control.pb.go` was not idempotent with the repo-canonical `make proto` toolchain. |
| `plan_cloud_G08_1.log` | `code_review_cloud_G08_1.log` | PASS | Generated proto header was aligned to `protoc v5.29.3`; temp generation diff, focused tests, and Control Plane-Edge wire smoke passed. |
## 구현/정리 내용
- Added the Control Plane-Edge status snapshot protobuf contract and generated Go type support.
- Added Control Plane active Edge client status request handling and tests.
- Added Edge connector status responses backed by the Edge service node snapshot surface and bootstrap wiring tests.
- Restored generated proto idempotence for `proto/gen/iop/control.pb.go` with the repo-canonical `protoc v5.29.3` toolchain.
## 최종 검증
- `PATH="/config/.local/bin:$PATH" make proto` - PASS; generated files are up to date with canonical local proto tooling.
- `PATH="/config/.local/bin:$PATH"; tmpdir=$(mktemp -d); protoc --go_out="$tmpdir" --go_opt=module=iop --proto_path=. proto/iop/runtime.proto proto/iop/node.proto proto/iop/control.proto proto/iop/job.proto && diff -u proto/gen/iop/control.pb.go "$tmpdir/proto/gen/iop/control.pb.go"; status=$?; rm -rf "$tmpdir"; exit $status` - PASS; no diff output.
- `git diff --name-only -- proto/gen/iop/node.pb.go proto/gen/iop/runtime.pb.go proto/gen/iop/job.pb.go` - PASS; no status-unrelated generated churn.
- `go test -count=1 ./apps/control-plane/internal/wire ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./proto/gen/...` - PASS; all tested packages passed and `proto/gen/iop` has no test files.
- `make test-control-plane-edge-wire` - PASS; markers included `CP: hello accepted`, `Edge: connected to control plane`, `CP: edge disconnected`, and `Control Plane-Edge wire smoke PASSED.`
## Roadmap Completion
- Milestone: `agent-roadmap/phase/control-plane-portal-ops/milestones/control-plane-edge-wire-baseline.md`
- Completed task ids:
- `edge-status`: PASS; evidence=`agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_0.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_0.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_1.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_1.log`; verification=`go test -count=1 ./apps/control-plane/internal/wire ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./proto/gen/...`, `make test-control-plane-edge-wire`
- Not completed task ids: 없음
## 잔여 Nit
- 없음
## 후속 작업
- 없음

0
agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_0.log → agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_0.log
View file

12
agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/PLAN-cloud-G08.md → agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_1.log
View file

@ -30,10 +30,10 @@
## 구현 체크리스트
- [ ] 현재 repo 툴체인으로 `make proto`를 실행해 `proto/gen/iop/control.pb.go`가 idempotent한 generated output이 되도록 맞춘다.
- [ ] repo 밖 temp dir에 Makefile과 동일한 protoc 명령을 실행해 `proto/gen/iop/control.pb.go`와 temp generated file의 diff가 없는지 확인하고, `node.pb.go`/`runtime.pb.go`/`job.pb.go`에 불필요한 diff가 남지 않았음을 기록한다.
- [ ] focused Go test와 Control Plane-Edge wire smoke를 재실행하고 실제 stdout/stderr를 review stub에 기록한다.
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
- [x] 현재 repo 툴체인으로 `make proto`를 실행해 `proto/gen/iop/control.pb.go`가 idempotent한 generated output이 되도록 맞춘다.
- [x] repo 밖 temp dir에 Makefile과 동일한 protoc 명령을 실행해 `proto/gen/iop/control.pb.go`와 temp generated file의 diff가 없는지 확인하고, `node.pb.go`/`runtime.pb.go`/`job.pb.go`에 불필요한 diff가 남지 않았음을 기록한다.
- [x] focused Go test와 Control Plane-Edge wire smoke를 재실행하고 실제 stdout/stderr를 review stub에 기록한다.
- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
## REVIEW_STATUS-1 Proto Generated Artifact Idempotence
@ -47,8 +47,8 @@
### 수정 파일 및 체크리스트
- [ ] `proto/gen/iop/control.pb.go`: 현재 repo `make proto` 출력과 idempotent하게 갱신.
- [ ] `CODE_REVIEW-cloud-G08.md`: 실제 `protoc --version`, `protoc-gen-go --version`, `make proto`, temp generation diff, 테스트/smoke 출력 기록.
- [x] `proto/gen/iop/control.pb.go`: 현재 repo `make proto` 출력과 idempotent하게 갱신.
- [x] `CODE_REVIEW-cloud-G08.md`: 실제 `protoc --version`, `protoc-gen-go --version`, `make proto`, temp generation diff, 테스트/smoke 출력 기록.
### 테스트 결정

401
agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_0.log Normal file
View file

@ -0,0 +1,401 @@
<!-- task=m-control-plane-edge-wire-baseline/05+04_event_relay_boundary plan=0 tag=EVENTS -->
# Code Review Reference - EVENTS
> **[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 user-only decision, user-owned external environment prerequisite, or scope conflict, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Evidence gaps that a follow-up agent can close by rerunning commands or collecting artifacts are normal follow-up issues, not user-review blockers by themselves.
> 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-05-30
task=m-control-plane-edge-wire-baseline/05+04_event_relay_boundary, plan=0, tag=EVENTS
## Roadmap Targets
- Milestone: `agent-roadmap/phase/control-plane-portal-ops/milestones/control-plane-edge-wire-baseline.md`
- Task ids:
- `event-relay`: Edge의 node lifecycle event가 Control Plane까지 relay되는 최소 경로가 있다.
- `service-boundary`: Edge wire handler는 node registry, run dispatch, command request 로직을 복제하지 않고 `apps/edge/internal/service`와 `events.Bus` 같은 표면 중립 경계를 사용한다.
- Completion mode: check-on-pass
## 이 파일을 읽는 리뷰 에이전트에게
> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다.
각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요.
리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다.
1. 판정을 append한다.
2. `CODE_REVIEW-cloud-G08.md` -> `code_review_cloud_G08_N.log`, `PLAN-cloud-G08.md` -> `plan_cloud_G08_M.log`로 아카이브한다.
3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다.
4. PASS이고 task group이 `m-control-plane-edge-wire-baseline`이므로 완료 이벤트 메타데이터를 보고한다. roadmap 상태 체크와 `update-roadmap` 호출은 런타임 책임이다.
5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다.
---
## 구현 항목별 완료 여부
| 항목 | 완료 여부 |
|------|---------|
| [EVENTS-1] Dependency Gate And Parser | [x] |
| [EVENTS-2] Control Plane Receive And Recent Event Store | [x] |
| [EVENTS-3] Edge Events.Bus Relay | [x] |
| [EVENTS-4] Observation Surface And Boundary Guard | [x] |
| [EVENTS-5] Regression And Smoke | [x] |
## 구현 체크리스트
- [x] `04_status_snapshot` subtask가 PASS 완료된 뒤에만 시작한다. runtime이 이 subtask를 시작했다면 의존성이 충족된 것으로 보고, 수동 실행이면 predecessor PASS evidence 또는 runtime 완료 이벤트 경로를 review stub에 기록한다.
- 완료 이벤트 경로: `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/complete.log`
- `agent-task/archive/**` 일반 열람 제한 때문에 archive 본문은 재열람하지 않고 완료 이벤트 경로만 기록했다.
- [x] Control Plane `EdgeParserMap`/`EdgeServer`가 `EdgeNodeEvent`를 수신해 edge id와 함께 bounded recent event store에 기록하도록 구현한다.
- [x] Edge `controlplane.Connector`가 `events.Bus.SubscribeAllNodes()`를 통해 node lifecycle event를 Control Plane으로 relay하도록 option과 bootstrap wiring을 추가한다.
- [x] Control Plane에서 relayed node lifecycle event를 조회할 최소 observation surface와 tests를 추가하되, Edge internal registry/run/command 로직을 복제하지 않는다.
- [x] Event relay, recent event snapshot, service/events boundary 회귀 테스트를 추가하고 대상 Go 테스트 및 기존 local wire smoke를 통과시킨다.
- [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_cloud_G08_N.log`로 아카이브한다.
- [x] active `PLAN-*-G??.md`를 `plan_cloud_G08_M.log`로 아카이브한다.
- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다.
- [ ] PASS이면 active task 디렉터리 `agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다.
- [ ] PASS이고 task group이 `m-control-plane-edge-wire-baseline`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-control-plane-edge-wire-baseline/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다.
- [x] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성하고 `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`를 남기지 않는다.
- [ ] USER_REVIEW가 사용자 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다.
## 계획 대비 변경 사항
- `GET /edges/{edge_id}/events`는 registry API의 `NodeEvents(edgeID, nodeID)` 형태에 맞춰 선택적 `node_id` query filter를 함께 지원한다. 기본 event 조회 표면은 계획대로 `/edges/{edge_id}/events`이다.
- `apps/edge/internal/controlplane/connector_test.go`의 기존 status 테스트 fixture가 `apps/edge/internal/node`를 직접 import하던 부분을 fake `StatusProvider`로 바꿨다. 구현 코드 변경은 아니지만 boundary guard `rg`가 test fixture까지 검사하므로 service/events boundary를 더 엄격하게 유지하기 위한 조정이다.
- 기존 local wire smoke는 계획대로 hello/connect/disconnect만 확인한다. event relay 성공 근거는 focused Go integration tests에 둔다.
## 주요 설계 결정
- Control Plane recent node events는 `EdgeRegistry` 안의 in-memory bounded buffer로 두고 상한은 128개로 고정했다. durable history, ack/retry, backpressure policy는 계획 범위 밖으로 유지했다.
- `RecordNodeEvent`와 `NodeEvents`는 `proto.Clone` 기반 copy를 사용해 caller mutation이 registry 내부 state를 오염시키지 않게 했다. 수신 event timestamp가 비어 있으면 Control Plane receive time으로 채운다.
- Edge connector relay는 hello accepted 이후 connection lifetime 동안만 `events.Bus.SubscribeAllNodes(64)`를 구독한다. relay 전 event copy에 `metadata["edge_id"]`와 누락 timestamp를 채우며 원본 bus event는 mutate하지 않는다.
- HTTP observation surface는 Control Plane에 도착한 recent event를 JSON으로 보여주는 데 한정하고, Edge node registry/run dispatch/command request 로직을 재구성하지 않는다.
## 사용자 리뷰 요청
_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._
- 상태: 없음
- 사유 유형: 없음
- 결정 필요: 없음
- 차단 근거: 없음
- 실행한 검증/명령: 없음
- 자동 후속 불가 이유: 없음
- 재개 조건: 없음
## 리뷰어를 위한 체크포인트
- `04_status_snapshot`의 predecessor PASS evidence 또는 runtime 완료 이벤트 경로가 없는 상태에서 코드 구현이 진행되지 않았는지 확인한다.
- Control Plane receive path가 connection의 `edgeID` context에 event를 묶고, hello 전 event를 무시하는지 확인한다.
- Edge connector relay가 `events.Bus.SubscribeAllNodes()`를 사용하고 node registry/run dispatch/command request 로직을 직접 복제하지 않는지 확인한다.
- relayed event copy/enrichment가 원본 bus event를 mutate하지 않는지 확인한다.
- `/edges/{edge_id}/events`는 Control Plane에 도착한 event 조회만 제공하며 Edge internals를 직접 재구성하지 않는지 확인한다.
- `-race` 검증과 review stub의 실제 stdout/stderr가 신뢰 가능한지 확인한다.
## 검증 결과
_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._
필수 규칙:
- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다.
- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다.
- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다.
- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다.
### EVENTS-1 중간 검증
```text
$ go test -count=1 ./apps/control-plane/internal/wire
ok iop/apps/control-plane/internal/wire 0.922s
```
### EVENTS-2 중간 검증
```text
$ go test -count=1 ./apps/control-plane/internal/wire
ok iop/apps/control-plane/internal/wire 0.922s
```
### EVENTS-3 중간 검증
```text
$ go test -count=1 ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events
ok iop/apps/edge/internal/controlplane 4.434s
ok iop/apps/edge/internal/bootstrap 0.013s
ok iop/apps/edge/internal/events 0.003s
```
### EVENTS-4 중간 검증
```text
$ go test -count=1 ./apps/control-plane/cmd/control-plane ./apps/control-plane/internal/wire
ok iop/apps/control-plane/cmd/control-plane 0.007s
ok iop/apps/control-plane/internal/wire 0.921s
$ rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane
(no matches; rg exit code 1)
```
### EVENTS-5 중간 검증
```text
$ go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire
ok iop/apps/edge/internal/controlplane 5.447s
ok iop/apps/control-plane/internal/wire 1.940s
$ go test ./...
ok iop/apps/control-plane/cmd/control-plane (cached)
ok iop/apps/control-plane/internal/wire 0.921s
ok iop/apps/edge/cmd/edge (cached)
ok iop/apps/edge/internal/bootstrap (cached)
ok iop/apps/edge/internal/controlplane 4.434s
ok iop/apps/edge/internal/events (cached)
ok iop/apps/edge/internal/input (cached)
ok iop/apps/edge/internal/input/a2a (cached)
ok iop/apps/edge/internal/node (cached)
ok iop/apps/edge/internal/openai (cached)
ok iop/apps/edge/internal/opsconsole (cached)
ok iop/apps/edge/internal/service (cached)
ok iop/apps/edge/internal/transport (cached)
ok iop/apps/node/cmd/node (cached)
ok iop/apps/node/internal/adapters (cached)
ok iop/apps/node/internal/adapters/cli (cached)
? iop/apps/node/internal/adapters/cli/internal/testutil [no test files]
ok iop/apps/node/internal/adapters/cli/status (cached)
? iop/apps/node/internal/adapters/mock [no test files]
ok iop/apps/node/internal/adapters/ollama (cached)
ok iop/apps/node/internal/adapters/vllm (cached)
ok iop/apps/node/internal/bootstrap (cached)
ok iop/apps/node/internal/node (cached)
ok iop/apps/node/internal/router (cached)
? iop/apps/node/internal/runtime [no test files]
ok iop/apps/node/internal/store (cached)
ok iop/apps/node/internal/transport (cached)
? iop/apps/worker/cmd/worker [no test files]
? iop/packages/auth [no test files]
ok iop/packages/config (cached)
? iop/packages/events [no test files]
ok iop/packages/hostsetup (cached)
? iop/packages/jobs [no test files]
? iop/packages/metadata [no test files]
ok iop/packages/observability (cached)
? iop/packages/policy [no test files]
? iop/packages/version [no test files]
? iop/proto/gen/iop [no test files]
$ make test-control-plane-edge-wire
./scripts/e2e-control-plane-edge-wire.sh
[cp-edge-wire] NOTE: auxiliary smoke only - verifies Control Plane-Edge hello and disconnect via real processes.
[cp-edge-wire] shellcheck not found, skipping
[cp-edge-wire] ports: cp_http=30079 cp_ws=30473 cp_edge_wire=31890 edge_node=32304 edge_bootstrap=33601 edge_metrics=34455
[cp-edge-wire] building temp binaries...
[cp-edge-wire] starting Control Plane...
[cp-edge-wire] waiting for Control Plane edge wire port 31890 (timeout: 20s)...
[cp-edge-wire] Control Plane edge wire port ready
[cp-edge-wire] starting Edge...
[cp-edge-wire] waiting for hello accepted (timeout: 30s)...
[cp-edge-wire] CP: hello accepted
[cp-edge-wire] Edge: connected to control plane
[cp-edge-wire] stopping Edge process to trigger disconnect...
[cp-edge-wire] waiting for disconnect marker on CP (timeout: 20s)...
[cp-edge-wire] CP: edge disconnected
=== CONTROL PLANE OUTPUT ===
{"level":"info","ts":1780146134.9294624,"caller":"control-plane/main.go:148","msg":"control-plane client wire endpoint reserved","protocol":"protobuf-socket","transport":"proto-socket-ws","listen":"127.0.0.1:30473"}
{"level":"info","ts":1780146134.9295163,"caller":"control-plane/main.go:153","msg":"control-plane edge wire endpoint reserved","protocol":"protobuf-socket","transport":"proto-socket-tcp","listen":"127.0.0.1:31890"}
{"level":"info","ts":1780146134.929524,"caller":"control-plane/main.go:159","msg":"control-plane database configured","configured":true,"host":"code-server-postgres:5432","database":"iop-control-plane-local"}
{"level":"info","ts":1780146134.9295301,"caller":"control-plane/main.go:162","msg":"control-plane redis configured","configured":true,"host":"code-server-redis:6379","database":"1","key_prefix":"iop:control-plane:local:"}
{"level":"info","ts":1780146134.9295626,"caller":"wire/client.go:92","msg":"starting client wire WS server","host":"127.0.0.1","port":30473,"path":"/client"}
{"level":"info","ts":1780146134.9307566,"caller":"wire/edge_server.go:143","msg":"starting edge wire TCP server","host":"127.0.0.1","port":31890,"transport":"proto-socket-tcp"}
{"level":"info","ts":1780146134.9308002,"caller":"control-plane/main.go:191","msg":"control-plane http endpoint listening","listen":"127.0.0.1:30079"}
{"level":"info","ts":1780146135.4408884,"caller":"wire/edge_server.go:104","msg":"edge hello accepted","edge_id":"smoke-edge-wire","edge_name":"Smoke Edge Wire","version":"0.1.0"}
{"level":"info","ts":1780146135.9413812,"caller":"wire/edge_server.go:129","msg":"edge disconnected","edge_id":"smoke-edge-wire","reason":"remote_closed"}
=== EDGE PROCESS OUTPUT ===
[Fx] PROVIDE fx.Lifecycle <= go.uber.org/fx.New.func1()
[Fx] PROVIDE fx.Shutdowner <= go.uber.org/fx.(*App).shutdowner-fm()
[Fx] PROVIDE fx.DotGraph <= go.uber.org/fx.(*App).dotGraph-fm()
[Fx] PROVIDE *config.EdgeConfig <= iop/apps/edge/internal/bootstrap.Module.func1()
[Fx] PROVIDE *bootstrap.Runtime <= iop/apps/edge/internal/bootstrap.NewRuntime()
[Fx] INVOKE iop/apps/edge/internal/bootstrap.Module.func2()
[Fx] RUN provide: go.uber.org/fx.New.func1()
[Fx] RUN provide: iop/apps/edge/internal/bootstrap.Module.func1()
[Fx] RUN provide: iop/apps/edge/internal/bootstrap.NewRuntime()
[Fx] HOOK OnStart iop/apps/edge/internal/bootstrap.Module.func2.1() executing (caller: iop/apps/edge/internal/bootstrap.Module.func2)
[Fx] HOOK OnStart iop/apps/edge/internal/bootstrap.Module.func2.1() called by iop/apps/edge/internal/bootstrap.Module.func2 ran successfully in 1.749166ms
[Fx] RUNNING
[Fx] TERMINATED
[Fx] HOOK OnStop iop/apps/edge/internal/bootstrap.Module.func2.2() executing (caller: iop/apps/edge/internal/bootstrap.Module.func2)
[Fx] HOOK OnStop iop/apps/edge/internal/bootstrap.Module.func2.2() called by iop/apps/edge/internal/bootstrap.Module.func2 ran successfully in 4.620333ms
=== EDGE LOG ===
{"level":"info","ts":1780146135.4399495,"caller":"transport/server.go:83","msg":"edge listening for nodes","addr":"127.0.0.1:32304"}
{"level":"warn","ts":1780146135.4400084,"logger":"bootstrap","caller":"bootstrap/artifact_server.go:35","msg":"bootstrap artifact directory does not exist","dir":"/tmp/tmp.zdXuWBIG0L/artifacts"}
{"level":"info","ts":1780146135.440279,"logger":"bootstrap","caller":"bootstrap/artifact_server.go:59","msg":"bootstrap artifact server listening","addr":"127.0.0.1:33601","dir":"/tmp/tmp.zdXuWBIG0L/artifacts"}
{"level":"info","ts":1780146135.441073,"logger":"controlplane","caller":"controlplane/connector.go:273","msg":"connected to control plane","wire_addr":"127.0.0.1:31890","protocol":"protobuf-socket"}
===========================
[cp-edge-wire] Control Plane-Edge wire smoke PASSED.
```
### 최종 검증
```text
$ go test -count=1 ./apps/control-plane/internal/wire
ok iop/apps/control-plane/internal/wire 0.922s
$ go test -count=1 ./apps/control-plane/cmd/control-plane
ok iop/apps/control-plane/cmd/control-plane 0.006s
$ go test -count=1 ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events
ok iop/apps/edge/internal/controlplane 4.434s
ok iop/apps/edge/internal/bootstrap 0.013s
ok iop/apps/edge/internal/events 0.003s
$ go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire
ok iop/apps/edge/internal/controlplane 5.447s
ok iop/apps/control-plane/internal/wire 1.940s
$ go test ./...
ok iop/apps/control-plane/cmd/control-plane (cached)
ok iop/apps/control-plane/internal/wire 0.921s
ok iop/apps/edge/cmd/edge (cached)
ok iop/apps/edge/internal/bootstrap (cached)
ok iop/apps/edge/internal/controlplane 4.434s
ok iop/apps/edge/internal/events (cached)
ok iop/apps/edge/internal/input (cached)
ok iop/apps/edge/internal/input/a2a (cached)
ok iop/apps/edge/internal/node (cached)
ok iop/apps/edge/internal/openai (cached)
ok iop/apps/edge/internal/opsconsole (cached)
ok iop/apps/edge/internal/service (cached)
ok iop/apps/edge/internal/transport (cached)
ok iop/apps/node/cmd/node (cached)
ok iop/apps/node/internal/adapters (cached)
ok iop/apps/node/internal/adapters/cli (cached)
? iop/apps/node/internal/adapters/cli/internal/testutil [no test files]
ok iop/apps/node/internal/adapters/cli/status (cached)
? iop/apps/node/internal/adapters/mock [no test files]
ok iop/apps/node/internal/adapters/ollama (cached)
ok iop/apps/node/internal/adapters/vllm (cached)
ok iop/apps/node/internal/bootstrap (cached)
ok iop/apps/node/internal/node (cached)
ok iop/apps/node/internal/router (cached)
? iop/apps/node/internal/runtime [no test files]
ok iop/apps/node/internal/store (cached)
ok iop/apps/node/internal/transport (cached)
? iop/apps/worker/cmd/worker [no test files]
? iop/packages/auth [no test files]
ok iop/packages/config (cached)
? iop/packages/events [no test files]
ok iop/packages/hostsetup (cached)
? iop/packages/jobs [no test files]
? iop/packages/metadata [no test files]
ok iop/packages/observability (cached)
? iop/packages/policy [no test files]
? iop/packages/version [no test files]
? iop/proto/gen/iop [no test files]
$ make test-control-plane-edge-wire
./scripts/e2e-control-plane-edge-wire.sh
[cp-edge-wire] NOTE: auxiliary smoke only - verifies Control Plane-Edge hello and disconnect via real processes.
[cp-edge-wire] shellcheck not found, skipping
[cp-edge-wire] ports: cp_http=30079 cp_ws=30473 cp_edge_wire=31890 edge_node=32304 edge_bootstrap=33601 edge_metrics=34455
[cp-edge-wire] building temp binaries...
[cp-edge-wire] starting Control Plane...
[cp-edge-wire] waiting for Control Plane edge wire port 31890 (timeout: 20s)...
[cp-edge-wire] Control Plane edge wire port ready
[cp-edge-wire] starting Edge...
[cp-edge-wire] waiting for hello accepted (timeout: 30s)...
[cp-edge-wire] CP: hello accepted
[cp-edge-wire] Edge: connected to control plane
[cp-edge-wire] stopping Edge process to trigger disconnect...
[cp-edge-wire] waiting for disconnect marker on CP (timeout: 20s)...
[cp-edge-wire] CP: edge disconnected
=== CONTROL PLANE OUTPUT ===
{"level":"info","ts":1780146134.9294624,"caller":"control-plane/main.go:148","msg":"control-plane client wire endpoint reserved","protocol":"protobuf-socket","transport":"proto-socket-ws","listen":"127.0.0.1:30473"}
{"level":"info","ts":1780146134.9295163,"caller":"control-plane/main.go:153","msg":"control-plane edge wire endpoint reserved","protocol":"protobuf-socket","transport":"proto-socket-tcp","listen":"127.0.0.1:31890"}
{"level":"info","ts":1780146134.929524,"caller":"control-plane/main.go:159","msg":"control-plane database configured","configured":true,"host":"code-server-postgres:5432","database":"iop-control-plane-local"}
{"level":"info","ts":1780146134.9295301,"caller":"control-plane/main.go:162","msg":"control-plane redis configured","configured":true,"host":"code-server-redis:6379","database":"1","key_prefix":"iop:control-plane:local:"}
{"level":"info","ts":1780146134.9295626,"caller":"wire/client.go:92","msg":"starting client wire WS server","host":"127.0.0.1","port":30473,"path":"/client"}
{"level":"info","ts":1780146134.9307566,"caller":"wire/edge_server.go:143","msg":"starting edge wire TCP server","host":"127.0.0.1","port":31890,"transport":"proto-socket-tcp"}
{"level":"info","ts":1780146134.9308002,"caller":"control-plane/main.go:191","msg":"control-plane http endpoint listening","listen":"127.0.0.1:30079"}
{"level":"info","ts":1780146135.4408884,"caller":"wire/edge_server.go:104","msg":"edge hello accepted","edge_id":"smoke-edge-wire","edge_name":"Smoke Edge Wire","version":"0.1.0"}
{"level":"info","ts":1780146135.9413812,"caller":"wire/edge_server.go:129","msg":"edge disconnected","edge_id":"smoke-edge-wire","reason":"remote_closed"}
=== EDGE PROCESS OUTPUT ===
[Fx] PROVIDE fx.Lifecycle <= go.uber.org/fx.New.func1()
[Fx] PROVIDE fx.Shutdowner <= go.uber.org/fx.(*App).shutdowner-fm()
[Fx] PROVIDE fx.DotGraph <= go.uber.org/fx.(*App).dotGraph-fm()
[Fx] PROVIDE *config.EdgeConfig <= iop/apps/edge/internal/bootstrap.Module.func1()
[Fx] PROVIDE *bootstrap.Runtime <= iop/apps/edge/internal/bootstrap.NewRuntime()
[Fx] INVOKE iop/apps/edge/internal/bootstrap.Module.func2()
[Fx] RUN provide: go.uber.org/fx.New.func1()
[Fx] RUN provide: iop/apps/edge/internal/bootstrap.Module.func1()
[Fx] RUN provide: iop/apps/edge/internal/bootstrap.NewRuntime()
[Fx] HOOK OnStart iop/apps/edge/internal/bootstrap.Module.func2.1() executing (caller: iop/apps/edge/internal/bootstrap.Module.func2)
[Fx] HOOK OnStart iop/apps/edge/internal/bootstrap.Module.func2.1() called by iop/apps/edge/internal/bootstrap.Module.func2 ran successfully in 1.749166ms
[Fx] RUNNING
[Fx] TERMINATED
[Fx] HOOK OnStop iop/apps/edge/internal/bootstrap.Module.func2.2() executing (caller: iop/apps/edge/internal/bootstrap.Module.func2)
[Fx] HOOK OnStop iop/apps/edge/internal/bootstrap.Module.func2.2() called by iop/apps/edge/internal/bootstrap.Module.func2 ran successfully in 4.620333ms
=== EDGE LOG ===
{"level":"info","ts":1780146135.4399495,"caller":"transport/server.go:83","msg":"edge listening for nodes","addr":"127.0.0.1:32304"}
{"level":"warn","ts":1780146135.4400084,"logger":"bootstrap","caller":"bootstrap/artifact_server.go:35","msg":"bootstrap artifact directory does not exist","dir":"/tmp/tmp.zdXuWBIG0L/artifacts"}
{"level":"info","ts":1780146135.440279,"logger":"bootstrap","caller":"bootstrap/artifact_server.go:59","msg":"bootstrap artifact server listening","addr":"127.0.0.1:33601","dir":"/tmp/tmp.zdXuWBIG0L/artifacts"}
{"level":"info","ts":1780146135.441073,"logger":"controlplane","caller":"controlplane/connector.go:273","msg":"connected to control plane","wire_addr":"127.0.0.1:31890","protocol":"protobuf-socket"}
===========================
[cp-edge-wire] Control Plane-Edge wire smoke PASSED.
$ rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane
(no matches; rg exit code 1)
```
---
> **[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.
## 섹션 소유권
| 섹션 | 소유자 | 설명 |
|------|--------|------|
| 헤더 주석, 개요, 리뷰 에이전트 지시 | 스텁 생성 시 고정 | 구현 에이전트가 수정하거나 실행하지 않음 |
| Roadmap Targets | plan에서 복사해 스텁 생성 시 고정 | 구현 에이전트가 수정하지 않음 |
| 구현 항목별 완료 여부 | 구현 에이전트 | `[ ]` -> `[x]` 체크 |
| 구현 체크리스트 | 구현 에이전트 | `[ ]` -> `[x]` 체크, 마지막 항목 필수 |
| 코드리뷰 전용 체크리스트 | 리뷰 에이전트 | 구현 에이전트가 수정하지 않음 |
| 계획 대비 변경 사항, 주요 설계 결정 | 구현 에이전트 | 실제 내용으로 교체 |
| 사용자 리뷰 요청 | 구현 에이전트 | 기본 `상태: 없음` 유지 또는 차단 근거 기록 |
| 리뷰어를 위한 체크포인트 | 스텁 생성 시 고정 | 리뷰 초점 |
| 검증 결과 | 구현 에이전트 | 실제 stdout/stderr 기록 |
| 코드리뷰 결과 | 리뷰 에이전트 | 스텁에 포함하지 않음 |
## 코드리뷰 결과
- 종합 판정: FAIL
- 차원별 평가:
- Correctness: Fail - `EdgeNodeEvent` 수신 경로가 기존 same-`edge_id` 재연결 token invariant를 적용하지 않아 stale connection event를 기록할 수 있다.
- Completeness: Pass - 계획된 parser, receive/store, relay, observation surface, boundary guard 작업은 대부분 구현되어 있다.
- Test coverage: Fail - stale reconnect 후 오래된 connection의 `EdgeNodeEvent`가 무시되는 회귀 테스트가 없다.
- API contract: Fail - `EdgeServer`/`EdgeRegistry`의 현재 연결 token 계약은 disconnect와 active client에는 적용되지만 새 event record path에는 적용되지 않는다.
- Code quality: Pass - 새 코드 구조는 기존 boundary를 대체로 따른다.
- Plan deviation: Pass - 명시 계획 범위 안의 구현이다.
- Verification trust: Warn - 기록된 출력 형태는 계획과 맞지만, 리뷰 중 `agent-test/local/rules.md`의 로컬 테스트 금지 규칙 때문에 Go/smoke 명령을 로컬 재실행하지 않았다.
- 발견된 문제:
- Required: `apps/control-plane/internal/wire/edge_server.go:75`의 `EdgeNodeEvent` listener는 `edgeID`만 읽고 `apps/control-plane/internal/wire/edge_registry.go:128`의 `RecordNodeEvent`로 바로 저장한다. 기존 서버는 같은 `edge_id`가 재연결되면 connection `token`으로 stale disconnect와 active client cleanup을 무시하는데, 새 event path는 token을 확인하지 않아 오래된 첫 번째 connection이 두 번째 connection으로 supersede된 뒤에도 `/edges/{edge_id}/events`를 오염시킬 수 있다. Fix: listener에서 `edgeID`와 `token`을 함께 캡처하고, registry lock 안에서 현재 state token과 일치할 때만 기록하는 token-aware record method를 사용한다. 같은 `edge_id` 두 connection을 만든 뒤 stale connection event는 무시되고 current connection event만 저장되는 회귀 테스트를 추가한다.
- 다음 단계: FAIL 후속 `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성해 stale connection event guard와 검증 증거를 보완한다.

285
agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_1.log Normal file
View file

@ -0,0 +1,285 @@
<!-- task=m-control-plane-edge-wire-baseline/05+04_event_relay_boundary plan=1 tag=REVIEW_EVENTS -->
# Code Review Reference - REVIEW_EVENTS
> **[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 user-only decision, user-owned external environment prerequisite, or scope conflict, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Evidence gaps that a follow-up agent can close by rerunning commands or collecting artifacts are normal follow-up issues, not user-review blockers by themselves.
> 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-05-30
task=m-control-plane-edge-wire-baseline/05+04_event_relay_boundary, plan=1, tag=REVIEW_EVENTS
## Roadmap Targets
- Milestone: `agent-roadmap/phase/control-plane-portal-ops/milestones/control-plane-edge-wire-baseline.md`
- Task ids:
- `event-relay`: Edge의 node lifecycle event가 Control Plane까지 relay되는 최소 경로가 있다.
- `service-boundary`: Edge wire handler는 node registry, run dispatch, command request 로직을 복제하지 않고 `apps/edge/internal/service`와 `events.Bus` 같은 표면 중립 경계를 사용한다.
- Completion mode: check-on-pass
## 이 파일을 읽는 리뷰 에이전트에게
> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다.
각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요.
리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다.
1. 판정을 append한다.
2. `CODE_REVIEW-cloud-G08.md` -> `code_review_cloud_G08_N.log`, `PLAN-cloud-G08.md` -> `plan_cloud_G08_M.log`로 아카이브한다.
3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다.
4. PASS이고 task group이 `m-control-plane-edge-wire-baseline`이므로 완료 이벤트 메타데이터를 보고한다. roadmap 상태 체크와 `update-roadmap` 호출은 런타임 책임이다.
5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다.
---
## 구현 항목별 완료 여부
| 항목 | 완료 여부 |
|------|---------|
| [REVIEW_EVENTS-1] Stale Connection Event Guard | [x] |
## 구현 체크리스트
- [x] Control Plane `EdgeServer`/`EdgeRegistry`의 node event 기록 경로가 connection token을 확인해 superseded stale connection event를 무시하도록 수정한다.
- [x] stale same-`edge_id` reconnection regression test를 추가해 stale connection event는 저장되지 않고 current connection event만 저장되는지 검증한다.
- [x] 원격 테스트 환경 규칙에 맞춰 focused/final verification을 실행하고 실제 stdout/stderr를 review stub에 기록한다.
- 지정 원격 명령은 `ssh toki@toki-labs.com` 접속 후 `/config/workspace/iop` 경로가 없어 모두 실패했다.
- 동일 workspace인 현재 컨테이너 `/config/workspace/iop`에서 같은 검증 명령을 fallback으로 실행했고 모두 통과했다.
- [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_cloud_G08_N.log`로 아카이브한다.
- [x] active `PLAN-*-G??.md`를 `plan_cloud_G08_M.log`로 아카이브한다.
- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다.
- [x] PASS이면 active task 디렉터리 `agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다.
- [x] PASS이고 task group이 `m-control-plane-edge-wire-baseline`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-control-plane-edge-wire-baseline/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다.
- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성하고 `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`를 남기지 않는다.
- [ ] USER_REVIEW가 사용자 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다.
## 계획 대비 변경 사항
- `RecordNodeEvent(edgeID, event, receivedAt)`는 기존 테스트/seed call site 호환을 위해 유지하고, server receive path 전용으로 `RecordNodeEventFromConnection(edgeID, token, event, receivedAt) bool`을 추가했다.
- PLAN의 원격 검증 명령은 `ssh toki@toki-labs.com`에서 `/config/workspace/iop` 경로가 없어 실행 불가했다. 원격 실패 stdout/stderr를 기록하고, 현재 실행 중인 동일 workspace `/config/workspace/iop`에서 fallback 검증을 수행했다.
## 주요 설계 결정
- connection-origin event append는 registry lock 안에서 current `EdgeConnectionState.token`과 listener가 캡처한 token을 비교한 뒤 수행한다.
- clone/timestamp 보정/bounded recent buffer 처리는 기존 behavior를 유지하기 위해 private `recordNodeEvent` helper로 모았다.
- stale reconnect regression은 같은 `edge_id`로 `conn1` hello 후 `conn2` hello를 보내 current token을 교체한 뒤, `conn1` event가 저장되지 않고 `conn2` event만 저장되는지 검증한다.
## 사용자 리뷰 요청
_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._
- 상태: 없음
- 사유 유형: 없음
- 결정 필요: 없음
- 차단 근거: 없음
- 실행한 검증/명령: 없음
- 자동 후속 불가 이유: 없음
- 재개 조건: 없음
## 리뷰어를 위한 체크포인트
- `EdgeNodeEvent` receive path가 `edgeID`뿐 아니라 connection `token`도 확인하는지 확인한다.
- 같은 `edge_id` stale reconnect regression이 오래된 connection event를 무시하고 current connection event만 기록하는지 확인한다.
- 기존 `RecordNodeEvent` copy/timestamp/bounded buffer behavior와 HTTP event view가 깨지지 않았는지 확인한다.
- proto 계약과 Edge connector relay loop가 불필요하게 변경되지 않았는지 확인한다.
- 원격 테스트 규칙에 맞는 stdout/stderr가 실제로 기록되었는지 확인한다.
## 검증 결과
_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._
필수 규칙:
- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다.
- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다.
- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다.
- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다.
### REVIEW_EVENTS-1 중간 검증
```text
$ ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/internal/wire'
zsh:cd:1: no such file or directory: /config/workspace/iop
$ go test -count=1 ./apps/control-plane/internal/wire
ok iop/apps/control-plane/internal/wire 1.293s
```
### 최종 검증
```text
$ ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/internal/wire'
zsh:cd:1: no such file or directory: /config/workspace/iop
$ go test -count=1 ./apps/control-plane/internal/wire
ok iop/apps/control-plane/internal/wire 1.293s
$ ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/cmd/control-plane ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events'
zsh:cd:1: no such file or directory: /config/workspace/iop
$ go test -count=1 ./apps/control-plane/cmd/control-plane ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events
ok iop/apps/control-plane/cmd/control-plane 0.008s
ok iop/apps/edge/internal/controlplane 4.433s
ok iop/apps/edge/internal/bootstrap 0.013s
ok iop/apps/edge/internal/events 0.004s
$ ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire'
zsh:cd:1: no such file or directory: /config/workspace/iop
$ go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire
ok iop/apps/edge/internal/controlplane 5.446s
ok iop/apps/control-plane/internal/wire 2.314s
$ ssh toki@toki-labs.com 'cd /config/workspace/iop && go test ./...'
zsh:cd:1: no such file or directory: /config/workspace/iop
$ go test ./...
ok iop/apps/control-plane/cmd/control-plane 0.007s
ok iop/apps/control-plane/internal/wire 1.295s
ok iop/apps/edge/cmd/edge (cached)
ok iop/apps/edge/internal/bootstrap (cached)
ok iop/apps/edge/internal/controlplane (cached)
ok iop/apps/edge/internal/events (cached)
ok iop/apps/edge/internal/input (cached)
ok iop/apps/edge/internal/input/a2a (cached)
ok iop/apps/edge/internal/node (cached)
ok iop/apps/edge/internal/openai (cached)
ok iop/apps/edge/internal/opsconsole (cached)
ok iop/apps/edge/internal/service (cached)
ok iop/apps/edge/internal/transport (cached)
ok iop/apps/node/cmd/node (cached)
ok iop/apps/node/internal/adapters (cached)
ok iop/apps/node/internal/adapters/cli (cached)
? iop/apps/node/internal/adapters/cli/internal/testutil [no test files]
ok iop/apps/node/internal/adapters/cli/status (cached)
? iop/apps/node/internal/adapters/mock [no test files]
ok iop/apps/node/internal/adapters/ollama (cached)
ok iop/apps/node/internal/adapters/vllm (cached)
ok iop/apps/node/internal/bootstrap (cached)
ok iop/apps/node/internal/node (cached)
ok iop/apps/node/internal/router (cached)
? iop/apps/node/internal/runtime [no test files]
ok iop/apps/node/internal/store (cached)
ok iop/apps/node/internal/transport (cached)
? iop/apps/worker/cmd/worker [no test files]
? iop/packages/auth [no test files]
ok iop/packages/config (cached)
? iop/packages/events [no test files]
ok iop/packages/hostsetup (cached)
? iop/packages/jobs [no test files]
? iop/packages/metadata [no test files]
ok iop/packages/observability (cached)
? iop/packages/policy [no test files]
? iop/packages/version [no test files]
? iop/proto/gen/iop [no test files]
$ ssh toki@toki-labs.com 'cd /config/workspace/iop && make test-control-plane-edge-wire'
zsh:cd:1: no such file or directory: /config/workspace/iop
$ make test-control-plane-edge-wire
./scripts/e2e-control-plane-edge-wire.sh
[cp-edge-wire] NOTE: auxiliary smoke only - verifies Control Plane-Edge hello and disconnect via real processes.
[cp-edge-wire] shellcheck not found, skipping
[cp-edge-wire] ports: cp_http=29879 cp_ws=30396 cp_edge_wire=32034 edge_node=32425 edge_bootstrap=33355 edge_metrics=35090
[cp-edge-wire] building temp binaries...
[cp-edge-wire] starting Control Plane...
[cp-edge-wire] waiting for Control Plane edge wire port 32034 (timeout: 20s)...
[cp-edge-wire] Control Plane edge wire port ready
[cp-edge-wire] starting Edge...
[cp-edge-wire] waiting for hello accepted (timeout: 30s)...
[cp-edge-wire] CP: hello accepted
[cp-edge-wire] Edge: connected to control plane
[cp-edge-wire] stopping Edge process to trigger disconnect...
[cp-edge-wire] waiting for disconnect marker on CP (timeout: 20s)...
[cp-edge-wire] CP: edge disconnected
=== CONTROL PLANE OUTPUT ===
{"level":"info","ts":1780147294.6073062,"caller":"control-plane/main.go:148","msg":"control-plane client wire endpoint reserved","protocol":"protobuf-socket","transport":"proto-socket-ws","listen":"127.0.0.1:30396"}
{"level":"info","ts":1780147294.607345,"caller":"control-plane/main.go:153","msg":"control-plane edge wire endpoint reserved","protocol":"protobuf-socket","transport":"proto-socket-tcp","listen":"127.0.0.1:32034"}
{"level":"info","ts":1780147294.6073515,"caller":"control-plane/main.go:159","msg":"control-plane database configured","configured":true,"host":"code-server-postgres:5432","database":"iop-control-plane-local"}
{"level":"info","ts":1780147294.607355,"caller":"control-plane/main.go:162","msg":"control-plane redis configured","configured":true,"host":"code-server-redis:6379","database":"1","key_prefix":"iop:control-plane:local:"}
{"level":"info","ts":1780147294.6073864,"caller":"wire/client.go:92","msg":"starting client wire WS server","host":"127.0.0.1","port":30396,"path":"/client"}
{"level":"info","ts":1780147294.6087651,"caller":"wire/edge_server.go:144","msg":"starting edge wire TCP server","host":"127.0.0.1","port":32034,"transport":"proto-socket-tcp"}
{"level":"info","ts":1780147294.6088116,"caller":"control-plane/main.go:191","msg":"control-plane http endpoint listening","listen":"127.0.0.1:29879"}
{"level":"info","ts":1780147295.117576,"caller":"wire/edge_server.go:105","msg":"edge hello accepted","edge_id":"smoke-edge-wire","edge_name":"Smoke Edge Wire","version":"0.1.0"}
{"level":"info","ts":1780147295.62148,"caller":"wire/edge_server.go:130","msg":"edge disconnected","edge_id":"smoke-edge-wire","reason":"remote_closed"}
=== EDGE PROCESS OUTPUT ===
[Fx] PROVIDE fx.Lifecycle <= go.uber.org/fx.New.func1()
[Fx] PROVIDE fx.Shutdowner <= go.uber.org/fx.(*App).shutdowner-fm()
[Fx] PROVIDE fx.DotGraph <= go.uber.org/fx.(*App).dotGraph-fm()
[Fx] PROVIDE *config.EdgeConfig <= iop/apps/edge/internal/bootstrap.Module.func1()
[Fx] PROVIDE *bootstrap.Runtime <= iop/apps/edge/internal/bootstrap.NewRuntime()
[Fx] INVOKE iop/apps/edge/internal/bootstrap.Module.func2()
[Fx] RUN provide: go.uber.org/fx.New.func1()
[Fx] RUN provide: iop/apps/edge/internal/bootstrap.Module.func1()
[Fx] RUN provide: iop/apps/edge/internal/bootstrap.NewRuntime()
[Fx] HOOK OnStart iop/apps/edge/internal/bootstrap.Module.func2.1() executing (caller: iop/apps/edge/internal/bootstrap.Module.func2)
[Fx] HOOK OnStart iop/apps/edge/internal/bootstrap.Module.func2.1() called by iop/apps/edge/internal/bootstrap.Module.func2 ran successfully in 2.07925ms
[Fx] RUNNING
[Fx] TERMINATED
[Fx] HOOK OnStop iop/apps/edge/internal/bootstrap.Module.func2.2() executing (caller: iop/apps/edge/internal/bootstrap.Module.func2)
[Fx] HOOK OnStop iop/apps/edge/internal/bootstrap.Module.func2.2() called by iop/apps/edge/internal/bootstrap.Module.func2 ran successfully in 2.08225ms
=== EDGE LOG ===
{"level":"info","ts":1780147295.1162636,"caller":"transport/server.go:83","msg":"edge listening for nodes","addr":"127.0.0.1:32425"}
{"level":"warn","ts":1780147295.1164565,"logger":"bootstrap","caller":"bootstrap/artifact_server.go:35","msg":"bootstrap artifact directory does not exist","dir":"/tmp/tmp.HXPUCQ8p5W/artifacts"}
{"level":"info","ts":1780147295.1168535,"logger":"bootstrap","caller":"bootstrap/artifact_server.go:59","msg":"bootstrap artifact server listening","addr":"127.0.0.1:33355","dir":"/tmp/tmp.HXPUCQ8p5W/artifacts"}
{"level":"info","ts":1780147295.1178906,"logger":"controlplane","caller":"controlplane/connector.go:273","msg":"connected to control plane","wire_addr":"127.0.0.1:32034","protocol":"protobuf-socket"}
===========================
[cp-edge-wire] Control Plane-Edge wire smoke PASSED.
$ ssh toki@toki-labs.com 'cd /config/workspace/iop && (rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane; test $? -eq 1)'
zsh:cd:1: no such file or directory: /config/workspace/iop
$ (rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane; test $? -eq 1)
(no output; exit code 0)
```
---
> **[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.
## 섹션 소유권
| 섹션 | 소유자 | 설명 |
|------|--------|------|
| 헤더 주석, 개요, 리뷰 에이전트 지시 | 스텁 생성 시 고정 | 구현 에이전트가 수정하거나 실행하지 않음 |
| Roadmap Targets | plan에서 복사해 스텁 생성 시 고정 | 구현 에이전트가 수정하지 않음 |
| 구현 항목별 완료 여부 | 구현 에이전트 | `[ ]` -> `[x]` 체크 |
| 구현 체크리스트 | 구현 에이전트 | `[ ]` -> `[x]` 체크, 마지막 항목 필수 |
| 코드리뷰 전용 체크리스트 | 리뷰 에이전트 | 구현 에이전트가 수정하지 않음 |
| 계획 대비 변경 사항, 주요 설계 결정 | 구현 에이전트 | 실제 내용으로 교체 |
| 사용자 리뷰 요청 | 구현 에이전트 | 기본 `상태: 없음` 유지 또는 차단 근거 기록 |
| 리뷰어를 위한 체크포인트 | 스텁 생성 시 고정 | 리뷰 초점 |
| 검증 결과 | 구현 에이전트 | 실제 stdout/stderr 기록 |
| 코드리뷰 결과 | 리뷰 에이전트 | 스텁에 포함하지 않음 |
## 코드리뷰 결과
- 종합 판정: PASS
- 차원별 평가:
- Correctness: Pass - connection-origin `EdgeNodeEvent` 기록이 current connection token과 일치할 때만 append되도록 보완되어 stale same-`edge_id` connection event 오염을 막는다.
- Completeness: Pass - follow-up 계획의 token-aware registry path, server listener 전달, stale reconnect regression test, review stub 기록이 완료되었다.
- Test coverage: Pass - `TestEdgeServerStaleConnectionNodeEventDoesNotPolluteReconnect`가 stale event reject와 current event accept를 검증한다.
- API contract: Pass - 기존 `RecordNodeEvent` seed/test call site는 유지하고, server receive path에만 `RecordNodeEventFromConnection`을 적용했다.
- Code quality: Pass - helper 분리가 기존 copy/timestamp/bounded buffer behavior를 유지한다.
- Plan deviation: Pass - 원격 `/config/workspace/iop` 경로 부재를 실제 stderr로 기록하고 현재 리뷰 대상 workspace에서 동일 명령 fallback을 실행한 점은 타당하게 기록되었다.
- Verification trust: Pass - 구현자는 원격 실패 stderr와 fallback 검증 stdout/stderr를 모두 기록했고, 리뷰에서도 원격 경로 부재와 관련 call site/boundary를 확인했다.
- 발견된 문제: 없음
- 다음 단계: PASS 종결. `complete.log` 작성 후 task directory를 archive로 이동한다.

49
agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/complete.log Normal file
View file

@ -0,0 +1,49 @@
# Complete - m-control-plane-edge-wire-baseline/05+04_event_relay_boundary
## 완료 일시
2026-05-30
## 요약
Control Plane-Edge node lifecycle event relay and service boundary completed after 2 review loops; final verdict PASS.
## 루프 이력
| Plan | Review | Verdict | 메모 |
|------|--------|---------|------|
| `plan_cloud_G08_0.log` | `code_review_cloud_G08_0.log` | FAIL | Event relay implementation was broadly complete, but Control Plane receive path accepted stale same-`edge_id` connection events without checking the active connection token. |
| `plan_cloud_G08_1.log` | `code_review_cloud_G08_1.log` | PASS | Token-aware event recording and stale reconnect regression test closed the Required issue. |
## 구현/정리 내용
- Added Control Plane parser/receive/store support for relayed `EdgeNodeEvent` messages and a bounded copied recent event buffer.
- Added Edge connector relay wiring from `events.Bus.SubscribeAllNodes()` through the Control Plane connection without importing Edge node registry/run/command internals.
- Added `/edges/{edge_id}/events` observation surface and tests for event JSON views.
- Added token-aware Control Plane event recording so stale superseded connections cannot pollute recent event snapshots.
## 최종 검증
- `ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/internal/wire'` - BLOCKED; remote shell reported `zsh:cd:1: no such file or directory: /config/workspace/iop`.
- `go test -count=1 ./apps/control-plane/internal/wire` - PASS; `ok iop/apps/control-plane/internal/wire`.
- `go test -count=1 ./apps/control-plane/cmd/control-plane ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events` - PASS; all listed packages passed.
- `go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire` - PASS; both packages passed under race detector.
- `go test ./...` - PASS; repository Go test suite passed with expected no-test packages.
- `make test-control-plane-edge-wire` - PASS; markers included `CP: hello accepted`, `Edge: connected to control plane`, `CP: edge disconnected`, and `Control Plane-Edge wire smoke PASSED.`
- `(rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane; test $? -eq 1)` - PASS; no boundary matches.
## Roadmap Completion
- Milestone: `agent-roadmap/phase/control-plane-portal-ops/milestones/control-plane-edge-wire-baseline.md`
- Completed task ids:
- `event-relay`: PASS; evidence=`agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_0.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_0.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_1.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_1.log`; verification=`go test -count=1 ./apps/control-plane/internal/wire`, `go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire`, `make test-control-plane-edge-wire`
- `service-boundary`: PASS; evidence=`agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_0.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_0.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_1.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_1.log`; verification=`(rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane; test $? -eq 1)`
- Not completed task ids: 없음
## 잔여 Nit
- 없음
## 후속 작업
- 없음

0
agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/PLAN-cloud-G08.md → agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_0.log
View file

179
agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_1.log Normal file
View file

@ -0,0 +1,179 @@
<!-- task=m-control-plane-edge-wire-baseline/05+04_event_relay_boundary plan=1 tag=REVIEW_EVENTS -->
# Review Follow-up Plan - Stale Edge Event Guard
## 이 파일을 읽는 구현 에이전트에게
`CODE_REVIEW-cloud-G08.md`의 구현 에이전트 소유 섹션을 채우는 것까지가 구현입니다. 구현 후 검증을 실행하고 실제 변경 내용, 검증 출력, 계획 대비 변경 사항을 review stub에 기록한 뒤 active 파일을 그대로 둔 채 리뷰 준비를 보고하세요. 최종 판정, 로그 rename, `complete.log`, archive 이동은 code-review 스킬 책임입니다.
구현 중 사용자만 결정할 수 있는 범위 변경, 사용자 소유 외부 환경/secret, 또는 현재 계획과 충돌하는 요구가 발견되면 `CODE_REVIEW-cloud-G08.md`의 `사용자 리뷰 요청` 섹션에 정확한 결정 항목과 근거를 기록하고 중단하세요. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해결할 수 있는 검증 증거 공백은 사용자 리뷰 요청이 아니라 일반 후속 작업입니다.
## 배경
1차 구현은 Control Plane-Edge node lifecycle event relay 경로를 추가했지만, 리뷰에서 stale connection event 오염 가능성이 발견되었습니다. 기존 `EdgeServer`는 같은 `edge_id` 재연결 시 connection token으로 stale disconnect와 active client cleanup을 무시합니다. 새 `EdgeNodeEvent` listener도 같은 token invariant를 따라야 Control Plane의 recent event view가 현재 연결 기준으로 유지됩니다.
## 사용자 리뷰 요청 흐름
구현 중 차단 사유가 생기면 active `CODE_REVIEW-cloud-G08.md`의 `사용자 리뷰 요청` 섹션을 `agent-ops/skills/common/_templates/implementation-user-review-request-section.md` 형식으로 채웁니다. code-review 스킬이 그 요청을 검증하고 실제 `USER_REVIEW.md` 작성 여부를 결정합니다.
## Roadmap Targets
- Milestone: `agent-roadmap/phase/control-plane-portal-ops/milestones/control-plane-edge-wire-baseline.md`
- Task ids:
- `event-relay`: Edge의 node lifecycle event가 Control Plane까지 relay되는 최소 경로가 있다.
- `service-boundary`: Edge wire handler는 node registry, run dispatch, command request 로직을 복제하지 않고 `apps/edge/internal/service`와 `events.Bus` 같은 표면 중립 경계를 사용한다.
- Completion mode: check-on-pass
## 분석 결과
### 읽은 파일
- `agent-ops/rules/project/rules.md`
- `agent-ops/rules/private/rules.md`
- `agent-ops/rules/common/rules-roadmap.md`
- `agent-roadmap/current.md`
- `agent-ops/rules/project/domain/control-plane/rules.md`
- `agent-ops/rules/project/domain/edge/rules.md`
- `agent-ops/rules/project/domain/platform-common/rules.md`
- `agent-ops/rules/project/domain/testing/rules.md`
- `agent-test/local/rules.md`
- `agent-test/local/control-plane-smoke.md`
- `agent-test/local/edge-smoke.md`
- `agent-test/local/platform-common-smoke.md`
- `agent-test/local/testing-smoke.md`
- `agent-ops/skills/common/router.md`
- `agent-ops/skills/common/code-review/SKILL.md`
- `agent-ops/skills/common/plan/SKILL.md`
- `agent-ops/skills/common/_templates/implementation-user-review-request-section.md`
- `agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_0.log`
- `agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_0.log`
- `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/complete.log`
- `apps/control-plane/internal/wire/edge.go`
- `apps/control-plane/internal/wire/edge_registry.go`
- `apps/control-plane/internal/wire/edge_server.go`
- `apps/control-plane/internal/wire/edge_server_test.go`
- `apps/control-plane/internal/wire/edge_test.go`
- `apps/control-plane/cmd/control-plane/main.go`
- `apps/control-plane/cmd/control-plane/main_test.go`
- `apps/edge/internal/controlplane/connector.go`
- `apps/edge/internal/controlplane/connector_test.go`
- `apps/edge/internal/bootstrap/runtime.go`
- `apps/edge/internal/bootstrap/runtime_test.go`
- `apps/edge/internal/events/bus.go`
### 테스트 환경 규칙
- `test_env=local`로 판단했고 `agent-test/local/rules.md`를 읽었습니다.
- 이 규칙은 로컬 테스트/검증을 금지하고 기본 테스트 환경을 `ssh toki@toki-labs.com`의 code-server 컨테이너로 둡니다.
- 매칭 profile은 `control-plane-smoke`, `edge-smoke`, `platform-common-smoke`, `testing-smoke`입니다.
- 후속 검증 명령은 원격 code-server 컨테이너에서 실행합니다. 원격 접근이 실패하면 exact command와 stdout/stderr를 review stub에 기록하고, user-owned 외부 환경 prerequisite인지 code-review가 판단할 수 있게 남깁니다.
- `proto/iop/*.proto` 원본 변경은 계획하지 않으므로 `make proto`는 필수가 아닙니다. proto 원본을 바꾸는 선택을 하면 범위 이탈로 기록하고 `make proto`와 생성물 diff 검증을 추가해야 합니다.
### 테스트 커버리지 공백
- stale same-`edge_id` reconnection 뒤 오래된 첫 번째 connection이 `EdgeNodeEvent`를 보내도 recent event store에 기록되지 않는지 검증하는 테스트가 없습니다.
- current connection의 event가 여전히 정상 저장되는지 같은 테스트에서 함께 확인해야 합니다.
### 심볼 참조
- renamed/removed symbol: none. 새 token-aware helper를 추가하거나 `RecordNodeEvent` 내부를 분리할 수 있지만 기존 public test helper 성격의 call site를 불필요하게 깨지 않도록 합니다.
### 분할 판단
- 이 follow-up은 기존 split subtask `05+04_event_relay_boundary` 안의 1차 리뷰 Required issue만 해결합니다.
- 새 subtask로 분리하지 않습니다. 변경 범위가 `EdgeServer`/`EdgeRegistry` receive path와 해당 regression test에 좁게 묶여 있고, 기존 event relay 작업 완료 판단을 보완하는 성격입니다.
- predecessor `04_status_snapshot`은 `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/complete.log`로 PASS 완료가 확인되었습니다.
### 범위 결정 근거
- Edge connector relay loop, Control Plane HTTP `/edges/{edge_id}/events` response shape, proto 계약, durable event history, ack/retry/backpressure policy는 변경하지 않습니다.
- 필요한 경우에만 `RecordNodeEvent` 구현을 private helper로 분리하고, server receive path에는 token-aware guard를 적용합니다.
- 검증 보완은 stale reconnect regression, focused wire package test, 기존 final verification 재확인에 한정합니다.
### 빌드 등급
- build lane: `cloud-G08`, review lane: `cloud-G08`.
- 근거: Control Plane-Edge wire concurrency/protocol invariant를 보완하는 follow-up이며, 기존 verification trust가 원격 실행 규칙 때문에 리뷰 재실행되지 않았습니다.
## 구현 체크리스트
- [ ] Control Plane `EdgeServer`/`EdgeRegistry`의 node event 기록 경로가 connection token을 확인해 superseded stale connection event를 무시하도록 수정한다.
- [ ] stale same-`edge_id` reconnection regression test를 추가해 stale connection event는 저장되지 않고 current connection event만 저장되는지 검증한다.
- [ ] 원격 테스트 환경 규칙에 맞춰 focused/final verification을 실행하고 실제 stdout/stderr를 review stub에 기록한다.
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
## REVIEW_EVENTS-1 Stale Connection Event Guard
### 문제
`apps/control-plane/internal/wire/edge_server.go:75`의 `EdgeNodeEvent` listener는 hello 이후 저장된 `edgeID`만 보고 `registry.RecordNodeEvent(id, event, time.Now())`를 호출합니다. 기존 `MarkDisconnected`와 active client cleanup은 connection `token`으로 stale reconnect를 방어하지만, `apps/control-plane/internal/wire/edge_registry.go:128`의 event 기록 경로에는 같은 guard가 없습니다.
결과적으로 같은 `edge_id`로 두 번째 connection이 supersede된 뒤에도 오래된 첫 번째 connection이 닫히지 않은 상태에서 event를 보내면 Control Plane recent event store가 stale event를 받아들일 수 있습니다.
### 해결 방법
`EdgeServer` listener에서 `edgeID`와 `token`을 함께 캡처하고, registry lock 안에서 현재 tracked state의 token과 일치할 때만 event를 append하는 token-aware path를 추가합니다. 기존 tests가 직접 seed 용도로 쓰는 `RecordNodeEvent(edgeID, event, receivedAt)`를 유지하려면 private helper를 분리하고 server path에서만 token guard variant를 사용합니다.
예상 형태:
```go
func (r *EdgeRegistry) RecordNodeEventFromConnection(edgeID string, token uint64, event *iop.EdgeNodeEvent, receivedAt time.Time) bool {
return r.recordNodeEvent(edgeID, token, true, event, receivedAt)
}
```
server listener:
```go
idMu.Lock()
id := edgeID
tok := token
idMu.Unlock()
if id == "" {
return
}
registry.RecordNodeEventFromConnection(id, tok, event, time.Now())
```
### 수정 파일 및 체크리스트
- [ ] `apps/control-plane/internal/wire/edge_registry.go`: token-aware record path 추가, 기존 copy/timestamp/bounded behavior 유지.
- [ ] `apps/control-plane/internal/wire/edge_server.go`: event listener가 `edgeID`와 `token`을 함께 전달하도록 수정.
- [ ] `apps/control-plane/internal/wire/edge_server_test.go`: stale reconnect event regression 추가.
### 테스트 작성
- `TestEdgeServerStaleConnectionNodeEventDoesNotPolluteReconnect` 또는 기존 `TestEdgeServerStaleDisconnectDoesNotClearReconnect` 확장.
- fixture: 같은 `edgeID`로 `conn1` hello, `conn2` hello를 순서대로 보낸다.
- assertion: `conn1.Send(EdgeNodeEvent{EventId:"evt-stale"})` 후 registry에 event가 없어야 한다. `conn2.Send(EdgeNodeEvent{EventId:"evt-current"})` 후 registry에는 current event만 있어야 한다.
### 중간 검증
```bash
ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/internal/wire'
```
기대 결과: stale reconnect event regression과 기존 wire tests가 통과합니다.
## 수정 파일 요약
| 파일 | 항목 |
|------|------|
| `apps/control-plane/internal/wire/edge_registry.go` | REVIEW_EVENTS-1 |
| `apps/control-plane/internal/wire/edge_server.go` | REVIEW_EVENTS-1 |
| `apps/control-plane/internal/wire/edge_server_test.go` | REVIEW_EVENTS-1 |
## 최종 검증
```bash
ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/internal/wire'
ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/cmd/control-plane ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events'
ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire'
ssh toki@toki-labs.com 'cd /config/workspace/iop && go test ./...'
ssh toki@toki-labs.com 'cd /config/workspace/iop && make test-control-plane-edge-wire'
ssh toki@toki-labs.com 'cd /config/workspace/iop && (rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane; test $? -eq 1)'
```
기대 결과: 모든 Go/race/smoke 명령이 성공하고, 마지막 boundary check는 match 없이 exit code 0 wrapper로 끝납니다.
모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다.

204
agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/CODE_REVIEW-cloud-G08.md
View file

@ -1,204 +0,0 @@
<!-- task=m-control-plane-edge-wire-baseline/05+04_event_relay_boundary plan=0 tag=EVENTS -->
# Code Review Reference - EVENTS
> **[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 user-only decision, user-owned external environment prerequisite, or scope conflict, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Evidence gaps that a follow-up agent can close by rerunning commands or collecting artifacts are normal follow-up issues, not user-review blockers by themselves.
> 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-05-30
task=m-control-plane-edge-wire-baseline/05+04_event_relay_boundary, plan=0, tag=EVENTS
## Roadmap Targets
- Milestone: `agent-roadmap/phase/control-plane-portal-ops/milestones/control-plane-edge-wire-baseline.md`
- Task ids:
- `event-relay`: Edge의 node lifecycle event가 Control Plane까지 relay되는 최소 경로가 있다.
- `service-boundary`: Edge wire handler는 node registry, run dispatch, command request 로직을 복제하지 않고 `apps/edge/internal/service`와 `events.Bus` 같은 표면 중립 경계를 사용한다.
- Completion mode: check-on-pass
## 이 파일을 읽는 리뷰 에이전트에게
> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다.
각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요.
리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다.
1. 판정을 append한다.
2. `CODE_REVIEW-cloud-G08.md` -> `code_review_cloud_G08_N.log`, `PLAN-cloud-G08.md` -> `plan_cloud_G08_M.log`로 아카이브한다.
3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다.
4. PASS이고 task group이 `m-control-plane-edge-wire-baseline`이므로 완료 이벤트 메타데이터를 보고한다. roadmap 상태 체크와 `update-roadmap` 호출은 런타임 책임이다.
5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다.
---
## 구현 항목별 완료 여부
| 항목 | 완료 여부 |
|------|---------|
| [EVENTS-1] Dependency Gate And Parser | [ ] |
| [EVENTS-2] Control Plane Receive And Recent Event Store | [ ] |
| [EVENTS-3] Edge Events.Bus Relay | [ ] |
| [EVENTS-4] Observation Surface And Boundary Guard | [ ] |
| [EVENTS-5] Regression And Smoke | [ ] |
## 구현 체크리스트
- [ ] `04_status_snapshot` subtask가 PASS 완료된 뒤에만 시작한다. runtime이 이 subtask를 시작했다면 의존성이 충족된 것으로 보고, 수동 실행이면 predecessor PASS evidence 또는 runtime 완료 이벤트 경로를 review stub에 기록한다.
- [ ] Control Plane `EdgeParserMap`/`EdgeServer`가 `EdgeNodeEvent`를 수신해 edge id와 함께 bounded recent event store에 기록하도록 구현한다.
- [ ] Edge `controlplane.Connector`가 `events.Bus.SubscribeAllNodes()`를 통해 node lifecycle event를 Control Plane으로 relay하도록 option과 bootstrap wiring을 추가한다.
- [ ] Control Plane에서 relayed node lifecycle event를 조회할 최소 observation surface와 tests를 추가하되, Edge internal registry/run/command 로직을 복제하지 않는다.
- [ ] Event relay, recent event snapshot, service/events boundary 회귀 테스트를 추가하고 대상 Go 테스트 및 기존 local wire smoke를 통과시킨다.
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
## 코드리뷰 전용 체크리스트
> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다.
> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다.
- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
- [ ] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다.
- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G08_N.log`로 아카이브한다.
- [ ] active `PLAN-*-G??.md`를 `plan_cloud_G08_M.log`로 아카이브한다.
- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다.
- [ ] PASS이면 active task 디렉터리 `agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다.
- [ ] PASS이고 task group이 `m-control-plane-edge-wire-baseline`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-control-plane-edge-wire-baseline/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다.
- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성하고 `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`를 남기지 않는다.
- [ ] USER_REVIEW가 사용자 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다.
## 계획 대비 변경 사항
_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._
## 주요 설계 결정
_구현 에이전트가 주요 설계 결정 사항을 기록한다._
## 사용자 리뷰 요청
_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._
- 상태: 없음
- 사유 유형: 없음
- 결정 필요: 없음
- 차단 근거: 없음
- 실행한 검증/명령: 없음
- 자동 후속 불가 이유: 없음
- 재개 조건: 없음
## 리뷰어를 위한 체크포인트
- `04_status_snapshot`의 predecessor PASS evidence 또는 runtime 완료 이벤트 경로가 없는 상태에서 코드 구현이 진행되지 않았는지 확인한다.
- Control Plane receive path가 connection의 `edgeID` context에 event를 묶고, hello 전 event를 무시하는지 확인한다.
- Edge connector relay가 `events.Bus.SubscribeAllNodes()`를 사용하고 node registry/run dispatch/command request 로직을 직접 복제하지 않는지 확인한다.
- relayed event copy/enrichment가 원본 bus event를 mutate하지 않는지 확인한다.
- `/edges/{edge_id}/events`는 Control Plane에 도착한 event 조회만 제공하며 Edge internals를 직접 재구성하지 않는지 확인한다.
- `-race` 검증과 review stub의 실제 stdout/stderr가 신뢰 가능한지 확인한다.
## 검증 결과
_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._
필수 규칙:
- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다.
- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다.
- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다.
- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다.
### EVENTS-1 중간 검증
```text
$ go test -count=1 ./apps/control-plane/internal/wire
(output)
```
### EVENTS-2 중간 검증
```text
$ go test -count=1 ./apps/control-plane/internal/wire
(output)
```
### EVENTS-3 중간 검증
```text
$ go test -count=1 ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events
(output)
```
### EVENTS-4 중간 검증
```text
$ go test -count=1 ./apps/control-plane/cmd/control-plane ./apps/control-plane/internal/wire
(output)
$ rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane
(output)
```
### EVENTS-5 중간 검증
```text
$ go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire
(output)
$ go test ./...
(output)
$ make test-control-plane-edge-wire
(output)
```
### 최종 검증
```text
$ go test -count=1 ./apps/control-plane/internal/wire
(output)
$ go test -count=1 ./apps/control-plane/cmd/control-plane
(output)
$ go test -count=1 ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events
(output)
$ go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire
(output)
$ go test ./...
(output)
$ make test-control-plane-edge-wire
(output)
$ rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane
(output)
```
---
> **[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.
## 섹션 소유권
| 섹션 | 소유자 | 설명 |
|------|--------|------|
| 헤더 주석, 개요, 리뷰 에이전트 지시 | 스텁 생성 시 고정 | 구현 에이전트가 수정하거나 실행하지 않음 |
| Roadmap Targets | plan에서 복사해 스텁 생성 시 고정 | 구현 에이전트가 수정하지 않음 |
| 구현 항목별 완료 여부 | 구현 에이전트 | `[ ]` -> `[x]` 체크 |
| 구현 체크리스트 | 구현 에이전트 | `[ ]` -> `[x]` 체크, 마지막 항목 필수 |
| 코드리뷰 전용 체크리스트 | 리뷰 에이전트 | 구현 에이전트가 수정하지 않음 |
| 계획 대비 변경 사항, 주요 설계 결정 | 구현 에이전트 | 실제 내용으로 교체 |
| 사용자 리뷰 요청 | 구현 에이전트 | 기본 `상태: 없음` 유지 또는 차단 근거 기록 |
| 리뷰어를 위한 체크포인트 | 스텁 생성 시 고정 | 리뷰 초점 |
| 검증 결과 | 구현 에이전트 | 실제 stdout/stderr 기록 |
| 코드리뷰 결과 | 리뷰 에이전트 | 스텁에 포함하지 않음 |

67
apps/control-plane/cmd/control-plane/main.go
View file

@ -220,6 +220,24 @@ type edgeRegistryView struct {
LastSeen time.Time `json:"last_seen"`
}
type edgeNodeEventsResponse struct {
EdgeID string `json:"edge_id"`
Events []edgeNodeEventView `json:"events"`
}
type edgeNodeEventView struct {
EdgeID string `json:"edge_id"`
EventID string `json:"event_id"`
Type string `json:"type"`
Source string `json:"source"`
NodeID string `json:"node_id"`
Alias string `json:"alias"`
Reason string `json:"reason"`
Timestamp time.Time `json:"timestamp"`
ReceivedAt time.Time `json:"received_at"`
Metadata map[string]string `json:"metadata,omitempty"`
}
func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistry) {
mux.HandleFunc("/edges", func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/edges" {
@ -244,11 +262,29 @@ func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistr
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
edgeID := strings.TrimPrefix(r.URL.Path, "/edges/")
if edgeID == "" || strings.Contains(edgeID, "/") {
edgePath := strings.TrimPrefix(r.URL.Path, "/edges/")
edgeID, rest, hasRest := strings.Cut(edgePath, "/")
if edgeID == "" {
http.NotFound(w, r)
return
}
if hasRest {
if rest != "events" {
http.NotFound(w, r)
return
}
if _, ok := registry.Snapshot(edgeID); !ok {
http.NotFound(w, r)
return
}
records := registry.NodeEvents(edgeID, r.URL.Query().Get("node_id"))
views := make([]edgeNodeEventView, 0, len(records))
for _, record := range records {
views = append(views, edgeNodeEventViewFromRecord(record))
}
writeJSON(w, http.StatusOK, edgeNodeEventsResponse{EdgeID: edgeID, Events: views})
return
}
state, ok := registry.Snapshot(edgeID)
if !ok {
http.NotFound(w, r)
@ -258,6 +294,33 @@ func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistr
})
}
func edgeNodeEventViewFromRecord(record wire.EdgeNodeEventRecord) edgeNodeEventView {
event := record.Event
view := edgeNodeEventView{
EdgeID: record.EdgeID,
ReceivedAt: record.ReceivedAt,
}
if event == nil {
return view
}
if event.GetTimestamp() != 0 {
view.Timestamp = time.Unix(0, event.GetTimestamp()).UTC()
}
view.EventID = event.GetEventId()
view.Type = event.GetType()
view.Source = event.GetSource()
view.NodeID = event.GetNodeId()
view.Alias = event.GetAlias()
view.Reason = event.GetReason()
if metadata := event.GetMetadata(); len(metadata) > 0 {
view.Metadata = make(map[string]string, len(metadata))
for k, v := range metadata {
view.Metadata[k] = v
}
}
return view
}
func edgeRegistryViewFromState(state wire.EdgeConnectionState) edgeRegistryView {
return edgeRegistryView{
EdgeID: state.EdgeID,

94
apps/control-plane/cmd/control-plane/main_test.go
View file

@ -298,6 +298,100 @@ func TestEdgeRegistryHTTPHandlersListAndGetEdge(t *testing.T) {
}
}
func TestEdgeRegistryHTTPHandlersListNodeEvents(t *testing.T) {
registry := wire.NewEdgeRegistry()
connectedAt := time.Unix(1780142200, 0).UTC()
registry.MarkConnected(&iop.EdgeHelloRequest{
EdgeId: "edge-a",
EdgeName: "Edge A",
Version: "1.0.0",
}, connectedAt)
registry.MarkConnected(&iop.EdgeHelloRequest{
EdgeId: "edge-b",
EdgeName: "Edge B",
Version: "1.0.0",
}, connectedAt)
eventAt := time.Unix(1780142300, 0).UTC()
receivedAt := eventAt.Add(time.Second)
registry.RecordNodeEvent("edge-a", &iop.EdgeNodeEvent{
EventId: "evt-node-1",
Type: "node.connected",
Source: "edge",
NodeId: "node-1",
Alias: "alpha",
Reason: "registered",
Timestamp: eventAt.UnixNano(),
Metadata: map[string]string{"rack": "r1"},
}, receivedAt)
registry.RecordNodeEvent("edge-b", &iop.EdgeNodeEvent{
EventId: "evt-node-2",
Type: "node.connected",
NodeId: "node-2",
Timestamp: eventAt.UnixNano(),
}, receivedAt)
mux := http.NewServeMux()
registerEdgeRegistryHandlers(mux, registry)
resp := httptest.NewRecorder()
mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events", nil))
if resp.Code != http.StatusOK {
t.Fatalf("GET /edges/edge-a/events status=%d body=%s", resp.Code, resp.Body.String())
}
var got edgeNodeEventsResponse
if err := json.Unmarshal(resp.Body.Bytes(), &got); err != nil {
t.Fatalf("decode events response: %v", err)
}
if got.EdgeID != "edge-a" {
t.Fatalf("edge_id: got %q want edge-a", got.EdgeID)
}
if len(got.Events) != 1 {
t.Fatalf("events len: got %d want 1", len(got.Events))
}
event := got.Events[0]
if event.EventID != "evt-node-1" || event.Type != "node.connected" {
t.Fatalf("unexpected event identity: %+v", event)
}
if event.NodeID != "node-1" || event.Alias != "alpha" || event.Reason != "registered" {
t.Fatalf("unexpected node event view: %+v", event)
}
if !event.Timestamp.Equal(eventAt) {
t.Fatalf("timestamp: got %s want %s", event.Timestamp, eventAt)
}
if !event.ReceivedAt.Equal(receivedAt) {
t.Fatalf("received_at: got %s want %s", event.ReceivedAt, receivedAt)
}
if event.Metadata["rack"] != "r1" {
t.Fatalf("metadata rack: got %q", event.Metadata["rack"])
}
filtered := httptest.NewRecorder()
mux.ServeHTTP(filtered, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events?node_id=missing", nil))
if filtered.Code != http.StatusOK {
t.Fatalf("filtered events status=%d body=%s", filtered.Code, filtered.Body.String())
}
var filteredGot edgeNodeEventsResponse
if err := json.Unmarshal(filtered.Body.Bytes(), &filteredGot); err != nil {
t.Fatalf("decode filtered events response: %v", err)
}
if len(filteredGot.Events) != 0 {
t.Fatalf("expected empty filtered events, got %+v", filteredGot.Events)
}
extraResp := httptest.NewRecorder()
mux.ServeHTTP(extraResp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events/extra", nil))
if extraResp.Code != http.StatusNotFound {
t.Fatalf("GET /edges/edge-a/events/extra status=%d body=%s", extraResp.Code, extraResp.Body.String())
}
missingResp := httptest.NewRecorder()
mux.ServeHTTP(missingResp, httptest.NewRequest(http.MethodGet, "/edges/missing-edge/events", nil))
if missingResp.Code != http.StatusNotFound {
t.Fatalf("GET /edges/missing-edge/events status=%d body=%s", missingResp.Code, missingResp.Body.String())
}
}
func TestEdgeRegistryHTTPHandlersRejectUnsupportedCases(t *testing.T) {
registry := wire.NewEdgeRegistry()
mux := http.NewServeMux()

4
apps/control-plane/internal/wire/edge.go
View file

@ -27,5 +27,9 @@ func EdgeParserMap() proto_socket.ParserMap {
res := &iop.EdgeStatusResponse{}
return res, proto.Unmarshal(b, res)
},
proto_socket.TypeNameOf(&iop.EdgeNodeEvent{}): func(b []byte) (proto.Message, error) {
event := &iop.EdgeNodeEvent{}
return event, proto.Unmarshal(b, event)
},
}
}

112
apps/control-plane/internal/wire/edge_registry.go
View file

@ -6,12 +6,14 @@ import (
"time"
proto_socket "git.toki-labs.com/toki/proto-socket/go"
"google.golang.org/protobuf/proto"
iop "iop/proto/gen/iop"
)
const recentNodeEventCapacity = 128
// EdgeConnectionState captures the Control Plane's internal view of a single
// Edge wire connection. It is a registry foundation only; public status query
// and event relay APIs are deferred to later subtasks.
// Edge wire connection.
type EdgeConnectionState struct {
EdgeID string
EdgeName string
@ -28,11 +30,20 @@ type EdgeConnectionState struct {
token uint64
}
// EdgeNodeEventRecord is the Control Plane's copied recent view of one node
// lifecycle event as received from a specific Edge connection.
type EdgeNodeEventRecord struct {
EdgeID string
Event *iop.EdgeNodeEvent
ReceivedAt time.Time
}
// EdgeRegistry stores Edge connection state keyed by edge_id behind a mutex.
type EdgeRegistry struct {
mu sync.Mutex
edges map[string]*EdgeConnectionState
nextToken uint64
mu sync.Mutex
edges map[string]*EdgeConnectionState
nextToken uint64
recentNodeEvents []EdgeNodeEventRecord
}
// NewEdgeRegistry returns an empty Edge connection registry.
@ -111,9 +122,100 @@ func (r *EdgeRegistry) Snapshots() []EdgeConnectionState {
return snapshots
}
// RecordNodeEvent stores a copied EdgeNodeEvent under the identified Edge. It
// is intended for tests and non-connection seed paths that do not have an
// active connection token.
func (r *EdgeRegistry) RecordNodeEvent(edgeID string, event *iop.EdgeNodeEvent, receivedAt time.Time) {
r.recordNodeEvent(edgeID, 0, false, event, receivedAt)
}
// RecordNodeEventFromConnection stores a copied EdgeNodeEvent only when the
// supplied token still identifies the current connection for edgeID. It returns
// false for stale superseded connections.
func (r *EdgeRegistry) RecordNodeEventFromConnection(edgeID string, token uint64, event *iop.EdgeNodeEvent, receivedAt time.Time) bool {
return r.recordNodeEvent(edgeID, token, true, event, receivedAt)
}
// recordNodeEvent appends to the bounded recent buffer. The recent buffer is
// global to the in-memory registry; durable event history is intentionally
// outside this baseline.
func (r *EdgeRegistry) recordNodeEvent(edgeID string, token uint64, requireToken bool, event *iop.EdgeNodeEvent, receivedAt time.Time) bool {
if edgeID == "" || event == nil {
return false
}
if receivedAt.IsZero() {
receivedAt = time.Now()
}
copied := cloneEdgeNodeEvent(event)
if copied.GetTimestamp() == 0 {
copied.Timestamp = receivedAt.UnixNano()
}
r.mu.Lock()
defer r.mu.Unlock()
if requireToken {
state, ok := r.edges[edgeID]
if !ok || state.token != token {
return false
}
}
if len(r.recentNodeEvents) >= recentNodeEventCapacity {
r.recentNodeEvents = append(r.recentNodeEvents[:0], r.recentNodeEvents[1:]...)
}
r.recentNodeEvents = append(r.recentNodeEvents, EdgeNodeEventRecord{
EdgeID: edgeID,
Event: copied,
ReceivedAt: receivedAt,
})
return true
}
// NodeEvents returns copied recent node lifecycle events for an Edge. When
// nodeID is non-empty, only events for that node are returned.
func (r *EdgeRegistry) NodeEvents(edgeID, nodeID string) []EdgeNodeEventRecord {
if edgeID == "" {
return nil
}
r.mu.Lock()
defer r.mu.Unlock()
out := make([]EdgeNodeEventRecord, 0, len(r.recentNodeEvents))
for _, record := range r.recentNodeEvents {
if record.EdgeID != edgeID {
continue
}
if nodeID != "" && record.Event.GetNodeId() != nodeID {
continue
}
out = append(out, cloneEdgeNodeEventRecord(record))
}
if len(out) == 0 {
return nil
}
return out
}
// Len returns the number of tracked Edge connections.
func (r *EdgeRegistry) Len() int {
r.mu.Lock()
defer r.mu.Unlock()
return len(r.edges)
}
func cloneEdgeNodeEventRecord(record EdgeNodeEventRecord) EdgeNodeEventRecord {
return EdgeNodeEventRecord{
EdgeID: record.EdgeID,
Event: cloneEdgeNodeEvent(record.Event),
ReceivedAt: record.ReceivedAt,
}
}
func cloneEdgeNodeEvent(event *iop.EdgeNodeEvent) *iop.EdgeNodeEvent {
if event == nil {
return nil
}
copied, ok := proto.Clone(event).(*iop.EdgeNodeEvent)
if !ok {
return &iop.EdgeNodeEvent{}
}
return copied
}

11
apps/control-plane/internal/wire/edge_server.go
View file

@ -72,6 +72,17 @@ func NewEdgeServer(listenAddr string, logger *zap.Logger) (*EdgeServer, error) {
token uint64
)
proto_socket.AddListenerTyped[*iop.EdgeNodeEvent](&client.Communicator, func(event *iop.EdgeNodeEvent) {
idMu.Lock()
id := edgeID
tok := token
idMu.Unlock()
if id == "" {
return
}
registry.RecordNodeEventFromConnection(id, tok, event, time.Now())
})
proto_socket.AddRequestListenerTyped(&client.Communicator, func(req *iop.EdgeHelloRequest) (*iop.EdgeHelloResponse, error) {
now := time.Now()
if req.GetEdgeId() == "" {

163
apps/control-plane/internal/wire/edge_server_test.go
View file

@ -225,6 +225,62 @@ func TestEdgeServerStaleDisconnectDoesNotClearReconnect(t *testing.T) {
}
}
func TestEdgeServerStaleConnectionNodeEventDoesNotPolluteReconnect(t *testing.T) {
server, port := startEdgeServer(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
const edgeID = "edge-dgx-group"
conn1 := dialEdge(t, ctx, port)
defer conn1.Close()
sendEdgeHello(t, conn1, edgeID)
conn2 := dialEdge(t, ctx, port)
defer conn2.Close()
sendEdgeHello(t, conn2, edgeID)
if err := conn1.Send(&iop.EdgeNodeEvent{
EventId: "evt-stale",
Type: "node.connected",
NodeId: "node-stale",
}); err != nil {
t.Fatalf("send stale EdgeNodeEvent: %v", err)
}
time.Sleep(300 * time.Millisecond)
if got := server.Registry().NodeEvents(edgeID, ""); len(got) != 0 {
t.Fatalf("stale connection event polluted registry: %+v", got)
}
if err := conn2.Send(&iop.EdgeNodeEvent{
EventId: "evt-current",
Type: "node.connected",
NodeId: "node-current",
}); err != nil {
t.Fatalf("send current EdgeNodeEvent: %v", err)
}
var records []EdgeNodeEventRecord
deadline := time.Now().Add(2 * time.Second)
for {
records = server.Registry().NodeEvents(edgeID, "")
if len(records) == 1 {
break
}
if time.Now().After(deadline) {
t.Fatalf("current node event was not recorded in time; got %d records", len(records))
}
time.Sleep(20 * time.Millisecond)
}
if records[0].Event.GetEventId() != "evt-current" {
t.Fatalf("recorded event_id: got %q want evt-current", records[0].Event.GetEventId())
}
if records[0].Event.GetNodeId() != "node-current" {
t.Fatalf("recorded node_id: got %q want node-current", records[0].Event.GetNodeId())
}
}
func TestEdgeServerRequestsStatusFromConnectedEdge(t *testing.T) {
server, port := startEdgeServer(t)
@ -270,6 +326,84 @@ func TestEdgeServerRequestsStatusFromConnectedEdge(t *testing.T) {
}
}
func TestEdgeServerRecordsNodeLifecycleEventFromEdge(t *testing.T) {
server, port := startEdgeServer(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
client := dialEdge(t, ctx, port)
defer client.Close()
if err := client.Send(&iop.EdgeNodeEvent{
EventId: "evt-before-hello",
Type: "node.connected",
NodeId: "node-early",
}); err != nil {
t.Fatalf("send pre-hello EdgeNodeEvent: %v", err)
}
time.Sleep(100 * time.Millisecond)
if got := server.Registry().NodeEvents("edge-dgx-group", ""); len(got) != 0 {
t.Fatalf("expected pre-hello node event to be ignored, got %+v", got)
}
sendEdgeHello(t, client, "edge-dgx-group")
event := &iop.EdgeNodeEvent{
EventId: "evt-node-1",
Type: "node.connected",
Source: "edge",
NodeId: "node-1",
Alias: "alpha",
Reason: "registered",
Metadata: map[string]string{"rack": "r1"},
}
if err := client.Send(event); err != nil {
t.Fatalf("send EdgeNodeEvent: %v", err)
}
var records []EdgeNodeEventRecord
deadline := time.Now().Add(2 * time.Second)
for {
records = server.Registry().NodeEvents("edge-dgx-group", "")
if len(records) == 1 {
break
}
if time.Now().After(deadline) {
t.Fatalf("node event was not recorded in time; got %d records", len(records))
}
time.Sleep(20 * time.Millisecond)
}
got := records[0]
if got.EdgeID != "edge-dgx-group" {
t.Errorf("edge_id: got %q want %q", got.EdgeID, "edge-dgx-group")
}
if got.Event.GetEventId() != "evt-node-1" || got.Event.GetType() != "node.connected" {
t.Fatalf("unexpected recorded event: %+v", got.Event)
}
if got.Event.GetNodeId() != "node-1" || got.Event.GetAlias() != "alpha" {
t.Fatalf("unexpected node identity: %+v", got.Event)
}
if got.Event.GetMetadata()["rack"] != "r1" {
t.Fatalf("metadata rack: got %q", got.Event.GetMetadata()["rack"])
}
if got.Event.GetTimestamp() == 0 {
t.Fatal("expected registry to fill missing event timestamp")
}
if event.GetTimestamp() != 0 {
t.Fatal("registry mutated original event timestamp")
}
records[0].Event.Metadata["rack"] = "mutated"
again := server.Registry().NodeEvents("edge-dgx-group", "node-1")
if len(again) != 1 {
t.Fatalf("filtered node events: got %d want 1", len(again))
}
if again[0].Event.GetMetadata()["rack"] != "r1" {
t.Fatalf("registry event was mutated through returned snapshot: %+v", again[0].Event.GetMetadata())
}
}
func TestEdgeServerRequestStatusFailsWhenEdgeNotConnected(t *testing.T) {
server, _ := startEdgeServer(t)
if _, err := server.RequestStatus("missing-edge", time.Second); err == nil {
@ -309,6 +443,35 @@ func TestEdgeRegistrySnapshotsAreSortedAndCopied(t *testing.T) {
}
}
func TestEdgeRegistryNodeEventsAreBoundedAndCopied(t *testing.T) {
registry := NewEdgeRegistry()
receivedAt := time.Unix(1780142400, 0).UTC()
for i := 0; i < recentNodeEventCapacity+2; i++ {
registry.RecordNodeEvent("edge-a", &iop.EdgeNodeEvent{
EventId: fmt.Sprintf("evt-%03d", i),
Type: "node.connected",
NodeId: "node-1",
Metadata: map[string]string{
"index": fmt.Sprintf("%d", i),
},
}, receivedAt.Add(time.Duration(i)*time.Nanosecond))
}
events := registry.NodeEvents("edge-a", "")
if len(events) != recentNodeEventCapacity {
t.Fatalf("node events len: got %d want %d", len(events), recentNodeEventCapacity)
}
if events[0].Event.GetEventId() != "evt-002" {
t.Fatalf("expected oldest retained event evt-002, got %q", events[0].Event.GetEventId())
}
events[0].Event.Metadata["index"] = "mutated"
again := registry.NodeEvents("edge-a", "")
if again[0].Event.GetMetadata()["index"] == "mutated" {
t.Fatal("NodeEvents returned mutable registry storage")
}
}
func TestEdgeHeartbeatWaitExceedsInterval(t *testing.T) {
if EdgeHeartbeatWaitSec <= EdgeHeartbeatIntervalSec {
t.Fatalf("EdgeHeartbeatWaitSec (%d) must exceed EdgeHeartbeatIntervalSec (%d)",

53
apps/control-plane/internal/wire/edge_test.go
View file

@ -172,6 +172,59 @@ func TestEdgeParserMapStatusRoundTrip(t *testing.T) {
}
}
func TestEdgeParserMapNodeEventRoundTrip(t *testing.T) {
parserMap := EdgeParserMap()
typeName := proto_socket.TypeNameOf(&iop.EdgeNodeEvent{})
parse, ok := parserMap[typeName]
if !ok {
t.Fatalf("missing parser for %s", typeName)
}
event := &iop.EdgeNodeEvent{
EventId: "evt-node-1",
Type: "node.connected",
Source: "edge",
NodeId: "node-1",
Alias: "alpha",
Reason: "registered",
Timestamp: time.Now().UnixNano(),
Metadata: map[string]string{
"rack": "r1",
},
}
payload, err := proto.Marshal(event)
if err != nil {
t.Fatalf("marshal EdgeNodeEvent: %v", err)
}
got, err := parse(payload)
if err != nil {
t.Fatalf("parse EdgeNodeEvent: %v", err)
}
parsed, ok := got.(*iop.EdgeNodeEvent)
if !ok {
t.Fatalf("expected *iop.EdgeNodeEvent, got %T", got)
}
if parsed.GetEventId() != event.GetEventId() {
t.Errorf("event_id: got %q want %q", parsed.GetEventId(), event.GetEventId())
}
if parsed.GetType() != event.GetType() {
t.Errorf("type: got %q want %q", parsed.GetType(), event.GetType())
}
if parsed.GetNodeId() != event.GetNodeId() {
t.Errorf("node_id: got %q want %q", parsed.GetNodeId(), event.GetNodeId())
}
if parsed.GetAlias() != event.GetAlias() {
t.Errorf("alias: got %q want %q", parsed.GetAlias(), event.GetAlias())
}
if parsed.GetTimestamp() != event.GetTimestamp() {
t.Errorf("timestamp: got %d want %d", parsed.GetTimestamp(), event.GetTimestamp())
}
if parsed.GetMetadata()["rack"] != "r1" {
t.Errorf("metadata rack: got %q want %q", parsed.GetMetadata()["rack"], "r1")
}
}
// TestEdgeStatusRequestDoesNotExposeNodeDirectScheduling guards that the status
// request stays a correlation-only contract and never grows direct Node
// addressing or scheduling fields.

1
apps/edge/internal/bootstrap/runtime.go
View file

@ -67,6 +67,7 @@ func NewRuntime(cfg *config.EdgeConfig) (*Runtime, error) {
version.Version,
logger.Named("controlplane"),
edgecontrolplane.WithStatusProvider(svc),
edgecontrolplane.WithNodeEventBus(bus),
)
return &Runtime{

13
apps/edge/internal/bootstrap/runtime_test.go
View file

@ -207,6 +207,19 @@ func TestNewRuntimeWiresStatusProviderIntoConnector(t *testing.T) {
}
}
// TestNewRuntimeWiresNodeEventBusIntoConnector verifies that node lifecycle
// events are relayed through the shared events.Bus boundary.
func TestNewRuntimeWiresNodeEventBusIntoConnector(t *testing.T) {
cfg := newTestConfig()
rt, err := NewRuntime(cfg)
if err != nil {
t.Fatalf("NewRuntime: %v", err)
}
if !rt.ControlPlane.NodeEventBusConfigured() {
t.Fatal("expected NewRuntime to wire the node event bus into the connector")
}
}
// TestRuntimeStartsControlPlaneConnectorWhenEnabled verifies that Start does
// not fail when a Control Plane connector is configured with a local fake
// server accepting hello. Uses a local TCP fake endpoint.

72
apps/edge/internal/controlplane/connector.go
View file

@ -13,6 +13,7 @@ import (
"go.uber.org/zap"
"google.golang.org/protobuf/proto"
edgeevents "iop/apps/edge/internal/events"
edgeservice "iop/apps/edge/internal/service"
"iop/packages/config"
iop "iop/proto/gen/iop"
@ -55,6 +56,13 @@ func WithStatusProvider(provider StatusProvider) Option {
return func(c *Connector) { c.statusProvider = provider }
}
// WithNodeEventBus wires the Edge lifecycle event fanout surface the Connector
// relays to Control Plane. It keeps the connector at the events.Bus boundary
// instead of reaching into node registry or transport internals.
func WithNodeEventBus(bus *edgeevents.Bus) Option {
return func(c *Connector) { c.nodeEvents = bus }
}
// Connector manages the outbound TCP connection from Edge to Control Plane.
// When disabled (enabled=false or empty wire_addr) it is a safe no-op.
type Connector struct {
@ -64,6 +72,7 @@ type Connector struct {
logger *zap.Logger
statusProvider StatusProvider
nodeEvents *edgeevents.Bus
state atomic.Int32 // stores State
cancelOnce sync.Once
@ -96,6 +105,11 @@ func (c *Connector) StatusProviderConfigured() bool {
return c.statusProvider != nil
}
// NodeEventBusConfigured reports whether node lifecycle event relay is wired.
func (c *Connector) NodeEventBusConfigured() bool {
return c.nodeEvents != nil
}
// Start begins the outbound connection loop. It is a no-op when enabled=false
// or wire_addr is empty. ctx cancellation is used only for the startup phase;
// the internal lifetime context drives reconnect. Stop() tears down the loop.
@ -252,6 +266,9 @@ func (c *Connector) connect(ctx context.Context) error {
default:
}
stopNodeRelay := c.startNodeEventRelay(ctx, cl)
defer stopNodeRelay()
c.state.Store(int32(StateConnected))
c.logger.Info("connected to control plane",
zap.String("wire_addr", c.cpConf.WireAddr),
@ -268,6 +285,61 @@ func (c *Connector) connect(ctx context.Context) error {
return nil
}
func (c *Connector) startNodeEventRelay(ctx context.Context, cl *toki.TcpClient) func() {
if c.nodeEvents == nil {
return func() {}
}
events, unsubscribe := c.nodeEvents.SubscribeAllNodes(64)
relayCtx, cancel := context.WithCancel(ctx)
go func() {
defer unsubscribe()
for {
select {
case <-relayCtx.Done():
return
case event, ok := <-events:
if !ok {
return
}
relay := c.nodeEventForRelay(event)
if relay == nil {
continue
}
if err := cl.Send(relay); err != nil {
c.logger.Warn("relay node lifecycle event to control plane failed",
zap.String("event_id", relay.GetEventId()),
zap.String("node_id", relay.GetNodeId()),
zap.Error(err),
)
}
}
}
}()
return cancel
}
func (c *Connector) nodeEventForRelay(event *iop.EdgeNodeEvent) *iop.EdgeNodeEvent {
if event == nil {
return nil
}
relay, ok := proto.Clone(event).(*iop.EdgeNodeEvent)
if !ok {
return nil
}
if relay.Metadata == nil {
relay.Metadata = make(map[string]string)
}
if c.edgeInfo.ID != "" {
if _, ok := relay.Metadata["edge_id"]; !ok {
relay.Metadata["edge_id"] = c.edgeInfo.ID
}
}
if relay.Timestamp == 0 {
relay.Timestamp = time.Now().UnixNano()
}
return relay
}
// buildStatusResponse converts the Edge-owned node snapshot surface into the
// status response. When no status provider is wired it returns an explicit
// error response rather than a silently empty snapshot.

136
apps/edge/internal/controlplane/connector_test.go
View file

@ -10,12 +10,20 @@ import (
toki "git.toki-labs.com/toki/proto-socket/go"
"google.golang.org/protobuf/proto"
edgenode "iop/apps/edge/internal/node"
edgeevents "iop/apps/edge/internal/events"
edgeservice "iop/apps/edge/internal/service"
"iop/packages/config"
iop "iop/proto/gen/iop"
)
type fakeStatusProvider struct {
nodes []edgeservice.NodeSnapshot
}
func (f fakeStatusProvider) ListNodeSnapshots() []edgeservice.NodeSnapshot {
return append([]edgeservice.NodeSnapshot(nil), f.nodes...)
}
// fakeCPParserMap returns a parser for EdgeHelloRequest on the server side.
func fakeCPParserMap() toki.ParserMap {
return toki.ParserMap{
@ -23,6 +31,10 @@ func fakeCPParserMap() toki.ParserMap {
m := &iop.EdgeHelloRequest{}
return m, proto.Unmarshal(b, m)
},
toki.TypeNameOf(&iop.EdgeNodeEvent{}): func(b []byte) (proto.Message, error) {
m := &iop.EdgeNodeEvent{}
return m, proto.Unmarshal(b, m)
},
}
}
@ -142,14 +154,46 @@ func startFakeCPServerForStatus(t *testing.T) (int, <-chan *toki.TcpClient, func
}
}
// TestConnectorRespondsToStatusRequestFromService verifies that a status
// request from the Control Plane is answered from the Edge service node
// snapshot surface rather than a direct registry/transport read.
func TestConnectorRespondsToStatusRequestFromService(t *testing.T) {
reg := edgenode.NewRegistry()
reg.Register(&edgenode.NodeEntry{NodeID: "node-1", Alias: "alpha"})
reg.Register(&edgenode.NodeEntry{NodeID: "node-2"})
svc := edgeservice.New(reg, nil)
func startFakeCPServerForNodeEvents(t *testing.T) (int, <-chan *iop.EdgeNodeEvent, func()) {
t.Helper()
port, err := freeTCPPort()
if err != nil {
t.Fatalf("startFakeCPServerForNodeEvents freeTCPPort: %v", err)
}
events := make(chan *iop.EdgeNodeEvent, 8)
srv := toki.NewTcpServer("127.0.0.1", port, func(conn net.Conn) *toki.TcpClient {
return toki.NewTcpClient(conn, 0, 0, fakeCPParserMap())
})
srv.OnClientConnected = func(client *toki.TcpClient) {
toki.AddRequestListenerTyped[*iop.EdgeHelloRequest, *iop.EdgeHelloResponse](
&client.Communicator,
func(_ *iop.EdgeHelloRequest) (*iop.EdgeHelloResponse, error) {
return &iop.EdgeHelloResponse{Accepted: true, Protocol: "iop/1"}, nil
},
)
toki.AddListenerTyped[*iop.EdgeNodeEvent](&client.Communicator, func(event *iop.EdgeNodeEvent) {
events <- event
})
}
ctx, cancel := context.WithCancel(context.Background())
if err := srv.Start(ctx); err != nil {
cancel()
t.Fatalf("startFakeCPServerForNodeEvents Start: %v", err)
}
return port, events, func() {
cancel()
_ = srv.Stop()
}
}
// TestConnectorRespondsToStatusRequestFromProvider verifies that a status
// request from the Control Plane is answered from the configured snapshot
// boundary rather than a direct registry/transport read.
func TestConnectorRespondsToStatusRequestFromProvider(t *testing.T) {
provider := fakeStatusProvider{nodes: []edgeservice.NodeSnapshot{
{NodeID: "node-1", Alias: "alpha", Label: "node0"},
{NodeID: "node-2", Label: "node1"},
}}
port, serverClients, stop := startFakeCPServerForStatus(t)
defer stop()
@ -163,7 +207,7 @@ func TestConnectorRespondsToStatusRequestFromService(t *testing.T) {
},
"0.2.0",
noopLogger(),
WithStatusProvider(svc),
WithStatusProvider(provider),
)
if !c.StatusProviderConfigured() {
t.Fatal("expected status provider to be configured")
@ -212,6 +256,78 @@ func TestConnectorRespondsToStatusRequestFromService(t *testing.T) {
}
}
func TestConnectorRelaysNodeLifecycleEventsToControlPlane(t *testing.T) {
bus := edgeevents.NewBus()
port, events, stop := startFakeCPServerForNodeEvents(t)
defer stop()
c := NewConnector(
config.EdgeInfo{ID: "edge-dgx", Name: "DGX Edge"},
config.EdgeControlPlaneConf{
Enabled: true,
WireAddr: wireAddr(port),
ReconnectIntervalSec: 60,
},
"0.2.0",
noopLogger(),
WithNodeEventBus(bus),
)
if !c.NodeEventBusConfigured() {
t.Fatal("expected node event bus to be configured")
}
if err := c.Start(context.Background()); err != nil {
t.Fatalf("Start: %v", err)
}
defer c.Stop()
deadline := time.Now().Add(3 * time.Second)
for c.CurrentState() != StateConnected {
if time.Now().After(deadline) {
t.Fatalf("connector did not reach connected state, got %v", c.CurrentState())
}
time.Sleep(20 * time.Millisecond)
}
original := &iop.EdgeNodeEvent{
EventId: "evt-node-1",
Type: "node.connected",
Source: "edge",
NodeId: "node-1",
Alias: "alpha",
Reason: "registered",
Metadata: map[string]string{"rack": "r1"},
}
bus.PublishNode(original)
select {
case got := <-events:
if got.GetEventId() != "evt-node-1" || got.GetType() != "node.connected" {
t.Fatalf("unexpected relayed event: %+v", got)
}
if got.GetNodeId() != "node-1" || got.GetAlias() != "alpha" {
t.Fatalf("unexpected relayed node identity: %+v", got)
}
if got.GetMetadata()["rack"] != "r1" {
t.Fatalf("metadata rack: got %q", got.GetMetadata()["rack"])
}
if got.GetMetadata()["edge_id"] != "edge-dgx" {
t.Fatalf("metadata edge_id: got %q want edge-dgx", got.GetMetadata()["edge_id"])
}
if got.GetTimestamp() == 0 {
t.Fatal("expected relay to fill missing timestamp on copied event")
}
case <-time.After(3 * time.Second):
t.Fatal("timeout waiting for relayed EdgeNodeEvent")
}
if _, ok := original.GetMetadata()["edge_id"]; ok {
t.Fatal("relay mutated original event metadata")
}
if original.GetTimestamp() != 0 {
t.Fatal("relay mutated original event timestamp")
}
}
// TestConnectorStatusRequestWithoutProviderReturnsError verifies that a status
// request is answered with an explicit error when no provider is wired.
func TestConnectorStatusRequestWithoutProviderReturnsError(t *testing.T) {

2
proto/gen/iop/control.pb.go
View file

@ -1,7 +1,7 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.11
// protoc v3.21.12
// protoc v5.29.3
// source: proto/iop/control.proto
package iop