From 41d2d4a92e0fb2d7027b63fdfdf0ee1aca1a4db0 Mon Sep 17 00:00:00 2001 From: toki Date: Tue, 9 Jun 2026 03:56:52 +0900 Subject: [PATCH] =?UTF-8?q?refactor(core):=20milestone=20=EC=A0=95?= =?UTF-8?q?=EB=A6=AC=EC=99=80=20=ED=95=B8=EB=93=A4=EB=9F=AC=20DTO=20?= =?UTF-8?q?=EA=B2=BD=EB=A1=9C=EB=A5=BC=20=EC=A0=95=EB=A6=AC=ED=95=9C?= =?UTF-8?q?=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 로드맵 milestone 상태를 archive로 이동하고 작업 메타데이터 추적을 정합성 있게 반영한다. core의 상태 저장 경로 변경 반영과 핸들러 파일 분리로 DTO 처리·라우트 의존을 정리했다. --- .../core-server-boundary-cleanup.md | 17 ++++- .../phase/independent-control-plane/PHASE.md | 6 +- .../workspace-metadata-reproducibility.md | 2 +- services/core/internal/cicdstate/store.go | 10 +++ .../core/internal/cicdstate/store_test.go | 19 +++++ services/core/internal/httpserver/dto.go | 74 +++++++++++++++++++ .../httpserver/runner_cicd_handlers.go | 73 +++--------------- .../internal/httpserver/runner_handlers.go | 69 +++-------------- 8 files changed, 141 insertions(+), 129 deletions(-) rename agent-roadmap/{ => archive}/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md (63%) diff --git a/agent-roadmap/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md b/agent-roadmap/archive/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md similarity index 63% rename from agent-roadmap/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md rename to agent-roadmap/archive/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md index 78de4a4..da28323 100644 --- a/agent-roadmap/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md +++ b/agent-roadmap/archive/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md @@ -1,7 +1,7 @@ ## 상태 -[진행중] +[완료] ## 목표 @@ -35,6 +35,17 @@ Go core service의 HTTP server, DTO 변환, runner/job handler, 시간 의존성 ### Epic: [http] server module boundary - [x] [routes] route 등록과 subrouter 구성을 handler 구현에서 분리한다. -- [ ] [handlers] runner, job/execution, log/artifact handler 책임을 기능별 파일 또는 내부 모듈로 나눈다. +- [x] [handlers] runner, job/execution, log/artifact handler 책임을 기능별 파일 또는 내부 모듈로 나눈다. - [x] [dto] HTTP DTO 변환과 domain state 조작 경계를 명확히 한다. -- [ ] [clock] timestamp 생성이 `Store`의 clock 주입 경로를 일관되게 사용하도록 정리한다. 검증: `make core-test`가 통과한다. +- [x] [clock] timestamp 생성이 `Store`의 clock 주입 경로를 일관되게 사용하도록 정리한다. 검증: `make core-test`가 통과한다. + +## 완료 리뷰 + +- 상태: 승인됨 +- 요청일: 2026-06-08 +- 승인일: 2026-06-09 +- 완료 근거: route 등록, handler 파일 분리, DTO 변환 경계가 `services/core/internal/httpserver/` 아래 책임별 파일로 정리되어 있다. +- 완료 근거: runner 계열 HTTP 응답 조립이 `dto.go` helper 경로를 사용하도록 정리되어 handler 내부의 직접 JSON map 응답 조립이 제거되었다. +- 완료 근거: runner job claim의 execution id 생성이 `cicdstate.Store`의 clock 경로를 사용하고, `CreateExecution`이 중복 execution id를 거부하도록 정리되었다. +- 완료 근거: `make core-test`가 통과했다. +- 사용자 최종 확인: 2026-06-09 완료 및 다음 마일스톤 전환 승인. diff --git a/agent-roadmap/phase/independent-control-plane/PHASE.md b/agent-roadmap/phase/independent-control-plane/PHASE.md index c0c372b..53373d2 100644 --- a/agent-roadmap/phase/independent-control-plane/PHASE.md +++ b/agent-roadmap/phase/independent-control-plane/PHASE.md @@ -30,10 +30,10 @@ OTO를 iop Edge에 직접 붙는 domain agent가 아니라, 독립 실행 가능 - [완료] oto_console 계약 경계 정리 - 경로: `agent-roadmap/archive/phase/independent-control-plane/milestones/oto-console-contract-boundary.md` - 요약: Flutter console contract와 shell 구현의 순환형 의존을 contract/model 소유 경계로 정리한다. -- [진행중] core server 모듈 경계 정리 - - 경로: `agent-roadmap/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md` +- [완료] core server 모듈 경계 정리 + - 경로: `agent-roadmap/archive/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md` - 요약: Go core service의 HTTP routing, handler, DTO, clock 의존을 모듈 경계별로 분리한다. -- [계획] workspace 메타데이터와 재현성 정리 +- [진행중] workspace 메타데이터와 재현성 정리 - 경로: `agent-roadmap/phase/independent-control-plane/milestones/workspace-metadata-reproducibility.md` - 요약: lockfile, local test 문서, package metadata, 배포 템플릿의 추적/설명 기준을 정리한다. diff --git a/agent-roadmap/phase/independent-control-plane/milestones/workspace-metadata-reproducibility.md b/agent-roadmap/phase/independent-control-plane/milestones/workspace-metadata-reproducibility.md index 212e710..3628629 100644 --- a/agent-roadmap/phase/independent-control-plane/milestones/workspace-metadata-reproducibility.md +++ b/agent-roadmap/phase/independent-control-plane/milestones/workspace-metadata-reproducibility.md @@ -2,7 +2,7 @@ ## 상태 -[계획] +[진행중] ## 목표 diff --git a/services/core/internal/cicdstate/store.go b/services/core/internal/cicdstate/store.go index aa1af3f..5aba4fc 100644 --- a/services/core/internal/cicdstate/store.go +++ b/services/core/internal/cicdstate/store.go @@ -109,6 +109,12 @@ func NewStore() *Store { } } +// NextExecutionID returns the next execution identifier using the store clock. +func (s *Store) NextExecutionID() string { + n := s.now() + return fmt.Sprintf("exec-%d", n.UnixNano()) +} + func (s *Store) CreateJob(id, name string, runInput *RunInput) (*Job, error) { s.mu.Lock() defer s.mu.Unlock() @@ -155,6 +161,10 @@ func (s *Store) CreateExecution(jobID, execID string) (*Execution, error) { return nil, fmt.Errorf("job %s already has execution %s", jobID, job.ExecutionID) } + if _, exists := s.executions[execID]; exists { + return nil, fmt.Errorf("execution %s already exists", execID) + } + now := s.now() exec := &Execution{ ID: execID, diff --git a/services/core/internal/cicdstate/store_test.go b/services/core/internal/cicdstate/store_test.go index 35eba99..62b4a9e 100644 --- a/services/core/internal/cicdstate/store_test.go +++ b/services/core/internal/cicdstate/store_test.go @@ -1,6 +1,7 @@ package cicdstate import ( + "fmt" "testing" "time" ) @@ -82,6 +83,24 @@ func TestStoreCreatesJobAndExecution(t *testing.T) { if err == nil { t.Fatal("expected error for second execution on same job") } + + store.CreateJob("job-2", "test", nil) + _, err = store.CreateExecution("job-2", "exec-1") + if err == nil { + t.Fatal("expected error for duplicate execution id") + } +} + +func TestStoreNextExecutionID(t *testing.T) { + store := NewStore() + fixed := time.Date(2026, 6, 5, 12, 0, 0, 0, time.UTC) + store.now = func() time.Time { return fixed } + + execID := store.NextExecutionID() + expected := fmt.Sprintf("exec-%d", fixed.UnixNano()) + if execID != expected { + t.Fatalf("execution id = %q, want %q", execID, expected) + } } func TestStoreCreatesJobWithRunInputCopy(t *testing.T) { diff --git a/services/core/internal/httpserver/dto.go b/services/core/internal/httpserver/dto.go index b4267cb..ce19d2b 100644 --- a/services/core/internal/httpserver/dto.go +++ b/services/core/internal/httpserver/dto.go @@ -8,6 +8,7 @@ import ( "time" "github.com/toki/oto/services/core/internal/cicdstate" + "github.com/toki/oto/services/core/internal/runnerregistry" otopb "github.com/toki/oto/services/core/oto" ) @@ -113,6 +114,79 @@ func artifactsToJSON(artifacts []cicdstate.ArtifactEntry) map[string]interface{} } } +func errorToJSON(message string) map[string]interface{} { + return map[string]interface{}{ + "error": message, + } +} + +func successToJSON(success bool) map[string]interface{} { + return map[string]interface{}{ + "success": success, + } +} + +func runnerDisconnectToJSON(success bool, err string) map[string]interface{} { + out := successToJSON(success) + if err != "" { + out["error"] = err + } + return out +} + +func runnerRecordToJSON(record runnerregistry.RunnerRecord) map[string]interface{} { + return map[string]interface{}{ + "runner_id": record.RunnerID, + "alias": record.Alias, + "protocol_version": record.ProtocolVersion, + "status": record.Status, + "accepted_at": record.AcceptedAt.Format(time.RFC3339), + "first_heartbeat_at": record.FirstHeartbeatAt.Format(time.RFC3339), + "last_heartbeat_at": record.LastHeartbeatAt.Format(time.RFC3339), + "failure_reason": record.FailureReason, + } +} + +func runnerStatusToJSON(record runnerregistry.RunnerRecord, runnerID string, lastExec *cicdstate.Execution) map[string]interface{} { + currentJobID := "" + currentExecID := "" + if lastExec != nil { + currentJobID = lastExec.JobID + currentExecID = lastExec.ID + } + return map[string]interface{}{ + "accepted": true, + "runner_id": runnerID, + "status": record.Status, + "current_job_id": currentJobID, + "current_execution_id": currentExecID, + "message": record.FailureReason, + } +} + +func runnerStatusRejectedToJSON(message string) map[string]interface{} { + return map[string]interface{}{ + "accepted": false, + "message": message, + } +} + +func selfUpdateToJSON(success bool, accepted bool, deferred bool, targetVersion string, message string) map[string]interface{} { + out := map[string]interface{}{ + "success": success, + "accepted": accepted, + "deferred": deferred, + "message": message, + } + if !success && !deferred { + out["error_message"] = message + } + if targetVersion != "" { + out["target_version"] = targetVersion + } + return out +} + func writeResponse(w http.ResponseWriter, status int, data interface{}) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) diff --git a/services/core/internal/httpserver/runner_cicd_handlers.go b/services/core/internal/httpserver/runner_cicd_handlers.go index 0dc81c7..1ff5713 100644 --- a/services/core/internal/httpserver/runner_cicd_handlers.go +++ b/services/core/internal/httpserver/runner_cicd_handlers.go @@ -5,7 +5,6 @@ import ( "fmt" "net/http" "strings" - "time" "github.com/toki/oto/services/core/internal/cicdstate" "github.com/toki/oto/services/core/internal/runnerregistry" @@ -58,7 +57,7 @@ func handleRunnerClaimJob(store *cicdstate.Store, registry *runnerregistry.Regis return } jobID = foundJob.ID - execID = fmt.Sprintf("exec-%d", time.Now().UnixNano()) + execID = store.NextExecutionID() } else if jobID == "" || execID == "" { writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ RunnerId: runnerID, @@ -66,6 +65,7 @@ func handleRunnerClaimJob(store *cicdstate.Store, registry *runnerregistry.Regis }) return } + job, err := store.GetJob(jobID) if err != nil { writeResponse(w, http.StatusNotFound, &otopb.JobClaimResponse{ @@ -380,10 +380,7 @@ func handleRunnerStatus(store *cicdstate.Store, registry *runnerregistry.Registr return func(w http.ResponseWriter, r *http.Request) { record, ok := registry.Snapshot(runnerID) if !ok { - writeResponse(w, http.StatusNotFound, map[string]interface{}{ - "accepted": false, - "message": "runner not found", - }) + writeResponse(w, http.StatusNotFound, runnerStatusRejectedToJSON("runner not found")) return } @@ -397,21 +394,7 @@ func handleRunnerStatus(store *cicdstate.Store, registry *runnerregistry.Registr } } - currentJobID := "" - currentExecID := "" - if lastExec != nil { - currentJobID = lastExec.JobID - currentExecID = lastExec.ID - } - - writeResponse(w, http.StatusOK, map[string]interface{}{ - "accepted": true, - "runner_id": runnerID, - "status": record.Status, - "current_job_id": currentJobID, - "current_execution_id": currentExecID, - "message": record.FailureReason, - }) + writeResponse(w, http.StatusOK, runnerStatusToJSON(record, runnerID, lastExec)) } } @@ -419,48 +402,24 @@ func handleRunnerSelfUpdate(store *cicdstate.Store, registry *runnerregistry.Reg return func(w http.ResponseWriter, r *http.Request) { var req otopb.SelfUpdateRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - writeResponse(w, http.StatusBadRequest, map[string]interface{}{ - "success": false, - "accepted": false, - "deferred": false, - "error_message": "invalid self-update request", - "message": "invalid self-update request", - }) + writeResponse(w, http.StatusBadRequest, selfUpdateToJSON(false, false, false, "", "invalid self-update request")) return } if req.RunnerId == "" { req.RunnerId = runnerID } else if req.RunnerId != runnerID { - writeResponse(w, http.StatusBadRequest, map[string]interface{}{ - "success": false, - "accepted": false, - "deferred": false, - "error_message": "runner id mismatch between path and body", - "message": "runner id mismatch between path and body", - }) + writeResponse(w, http.StatusBadRequest, selfUpdateToJSON(false, false, false, "", "runner id mismatch between path and body")) return } if err := ensureRunnerKnown(registry, runnerID); err != nil { - writeResponse(w, http.StatusNotFound, map[string]interface{}{ - "success": false, - "accepted": false, - "deferred": false, - "error_message": err.Error(), - "message": err.Error(), - }) + writeResponse(w, http.StatusNotFound, selfUpdateToJSON(false, false, false, "", err.Error())) return } if !strings.HasPrefix(req.GetDownloadUrl(), "https://") { - writeResponse(w, http.StatusBadRequest, map[string]interface{}{ - "success": false, - "accepted": false, - "deferred": false, - "error_message": "download_url must use HTTPS scheme", - "message": "download_url must use HTTPS scheme", - }) + writeResponse(w, http.StatusBadRequest, selfUpdateToJSON(false, false, false, "", "download_url must use HTTPS scheme")) return } @@ -473,22 +432,10 @@ func handleRunnerSelfUpdate(store *cicdstate.Store, registry *runnerregistry.Reg } if hasActiveExecution { - writeResponse(w, http.StatusOK, map[string]interface{}{ - "success": false, - "accepted": false, - "deferred": true, - "target_version": req.GetVersion(), - "message": "deferred: runner is currently running a job", - }) + writeResponse(w, http.StatusOK, selfUpdateToJSON(false, false, true, req.GetVersion(), "deferred: runner is currently running a job")) return } - writeResponse(w, http.StatusOK, map[string]interface{}{ - "success": true, - "accepted": true, - "deferred": false, - "target_version": req.GetVersion(), - "message": "self-update request accepted", - }) + writeResponse(w, http.StatusOK, selfUpdateToJSON(true, true, false, req.GetVersion(), "self-update request accepted")) } } diff --git a/services/core/internal/httpserver/runner_handlers.go b/services/core/internal/httpserver/runner_handlers.go index ac61cf0..6e0d11f 100644 --- a/services/core/internal/httpserver/runner_handlers.go +++ b/services/core/internal/httpserver/runner_handlers.go @@ -7,7 +7,6 @@ import ( "net/url" "os" "strings" - "time" "github.com/toki/oto/services/core/internal/runnerregistry" otopb "github.com/toki/oto/services/core/oto" @@ -130,30 +129,17 @@ func handleRunnerDisconnect(registry *runnerregistry.Registry) http.HandlerFunc } if runnerID == "" { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "success": false, - "error": "missing runner id", - }) + writeResponse(w, http.StatusBadRequest, runnerDisconnectToJSON(false, "missing runner id")) return } success := registry.Disconnect(runnerID) - w.Header().Set("Content-Type", "application/json") if !success { - w.WriteHeader(http.StatusNotFound) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "success": false, - "error": "runner not found", - }) + writeResponse(w, http.StatusNotFound, runnerDisconnectToJSON(false, "runner not found")) return } - w.WriteHeader(http.StatusOK) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "success": true, - }) + writeResponse(w, http.StatusOK, runnerDisconnectToJSON(true, "")) } } @@ -183,18 +169,7 @@ func handleGetRunner(registry *runnerregistry.Registry) http.HandlerFunc { return } - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusOK) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "runner_id": record.RunnerID, - "alias": record.Alias, - "protocol_version": record.ProtocolVersion, - "status": record.Status, - "accepted_at": record.AcceptedAt.Format(time.RFC3339), - "first_heartbeat_at": record.FirstHeartbeatAt.Format(time.RFC3339), - "last_heartbeat_at": record.LastHeartbeatAt.Format(time.RFC3339), - "failure_reason": record.FailureReason, - }) + writeResponse(w, http.StatusOK, runnerRecordToJSON(record)) } } @@ -211,22 +186,14 @@ func handleRunnerBootstrapCommand(registry *runnerregistry.Registry) http.Handle var request otopb.BootstrapCommandRequest if err := json.NewDecoder(r.Body).Decode(&request); err != nil { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "error": "invalid bootstrap command request", - }) + writeResponse(w, http.StatusBadRequest, errorToJSON("invalid bootstrap command request")) return } runnerID := strings.TrimSpace(request.GetRunnerId()) token := strings.TrimSpace(request.GetEnrollmentToken()) if runnerID == "" || token == "" { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "error": "missing runner id or enrollment token", - }) + writeResponse(w, http.StatusBadRequest, errorToJSON("missing runner id or enrollment token")) return } @@ -238,11 +205,7 @@ func handleRunnerBootstrapCommand(registry *runnerregistry.Registry) http.Handle u, err := url.Parse(serverURL) if err != nil || u.Host == "" || u.Scheme == "" { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "error": "invalid server URL", - }) + writeResponse(w, http.StatusBadRequest, errorToJSON("invalid server URL")) return } @@ -250,11 +213,7 @@ func handleRunnerBootstrapCommand(registry *runnerregistry.Registry) http.Handle for _, char := range u.Host { if !((char >= 'a' && char <= 'z') || (char >= 'A' && char <= 'Z') || (char >= '0' && char <= '9') || char == '.' || char == '-' || char == ':' || char == '[' || char == ']') { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "error": "invalid characters in server Host", - }) + writeResponse(w, http.StatusBadRequest, errorToJSON("invalid characters in server Host")) return } } @@ -264,20 +223,12 @@ func handleRunnerBootstrapCommand(registry *runnerregistry.Registry) http.Handle if scheme == "https" { releaseBaseURL = serverURL + "/releases" } else { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "error": "release base URL must use HTTPS (set OTO_RUNNER_RELEASE_BASE_URL environment variable)", - }) + writeResponse(w, http.StatusBadRequest, errorToJSON("release base URL must use HTTPS (set OTO_RUNNER_RELEASE_BASE_URL environment variable)")) return } } else { if !strings.HasPrefix(releaseBaseURL, "https://") { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "error": "OTO_RUNNER_RELEASE_BASE_URL must use https:// scheme", - }) + writeResponse(w, http.StatusBadRequest, errorToJSON("OTO_RUNNER_RELEASE_BASE_URL must use https:// scheme")) return } }