From e36db7281bebf08607ec14689706cb81f6855b22 Mon Sep 17 00:00:00 2001 From: toki Date: Sat, 30 May 2026 19:32:48 +0900 Subject: [PATCH] feat: update project rules, roadmap, and core service changes - Update agent-ops project rules and roadmap files - Add proto-socket infrastructure communication rail milestone - Update Flutter pubspec.lock and contracts notes - Enhance core service: config, HTTP middleware, router - Add notification module improvements - Add protosocket internal package --- agent-ops/rules/project/rules.md | 1 + agent-roadmap/ROADMAP.md | 2 +- agent-roadmap/current.md | 3 + agent-roadmap/phase/workflow-core/PHASE.md | 9 +- ...ocket-infrastructure-communication-rail.md | 93 ++++ .../workflow-core/milestones/workflow-core.md | 5 +- .../01_core_endpoint/USER_REVIEW.md | 80 ++++ apps/client/pubspec.lock | 4 +- .../notes/flutter-core-api-candidates.md | 145 +++++- services/core/cmd/server/main.go | 16 +- services/core/go.mod | 3 + services/core/go.sum | 8 + services/core/internal/config/config.go | 90 ++-- services/core/internal/config/config_test.go | 28 ++ services/core/internal/http/middleware.go | 13 + .../core/internal/http/middleware_test.go | 25 +- services/core/internal/http/router.go | 6 +- services/core/internal/notification/model.go | 8 + .../core/internal/notification/service.go | 28 +- .../internal/notification/service_test.go | 57 +++ .../core/internal/protosocket/dispatcher.go | 59 +++ .../core/internal/protosocket/envelope.go | 121 +++++ .../internal/protosocket/envelope_test.go | 116 +++++ services/core/internal/protosocket/events.go | 75 +++ services/core/internal/protosocket/server.go | 153 ++++++ .../core/internal/protosocket/server_test.go | 138 ++++++ services/core/internal/protosocket/tasks.go | 182 +++++++ .../core/internal/protosocket/tasks_test.go | 448 ++++++++++++++++++ 28 files changed, 1836 insertions(+), 80 deletions(-) create mode 100644 agent-roadmap/phase/workflow-core/milestones/proto-socket-infrastructure-communication-rail.md create mode 100644 agent-task/archive/2026/05/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/USER_REVIEW.md create mode 100644 services/core/internal/protosocket/dispatcher.go create mode 100644 services/core/internal/protosocket/envelope.go create mode 100644 services/core/internal/protosocket/envelope_test.go create mode 100644 services/core/internal/protosocket/events.go create mode 100644 services/core/internal/protosocket/server.go create mode 100644 services/core/internal/protosocket/server_test.go create mode 100644 services/core/internal/protosocket/tasks.go create mode 100644 services/core/internal/protosocket/tasks_test.go diff --git a/agent-ops/rules/project/rules.md b/agent-ops/rules/project/rules.md index 1240479..d1bd7a8 100644 --- a/agent-ops/rules/project/rules.md +++ b/agent-ops/rules/project/rules.md @@ -29,6 +29,7 @@ NomadCode는 AI 병렬 작업 운용을 위한 원레포다. 백엔드 오케스 - 원레포 루트에서 가능한 작업은 `bin/test`, `bin/lint`, `bin/build`, `bin/dev`를 우선 사용한다. - cross-boundary 변경은 계약, 서비스, 클라이언트 반영을 같은 브랜치에서 닫는다. +- 내부 런타임 간 신규 통신 규약은 `proto-socket`을 기본 표준으로 삼는다. REST/HTTP는 health/readiness, 외부 provider 호환 API, 운영 smoke/curl, public callback, 단순 정적 조회처럼 `proto-socket`이 과한 경우에만 예외로 쓰고, 예외 사유를 contracts 또는 관련 domain rule에 남긴다. - `services/core` 변경은 `go test ./...` 또는 `bin/test`로 확인한다. - `apps/client` 변경은 `flutter test`와 필요 시 `flutter analyze --no-fatal-infos`로 확인한다. - `packages/contracts`가 실제 계약을 갖기 전에는 API 형태를 서비스/클라이언트에 중복 고정하지 말고 계약 후보를 문서화한다. diff --git a/agent-roadmap/ROADMAP.md b/agent-roadmap/ROADMAP.md index ab9a326..356fe7a 100644 --- a/agent-roadmap/ROADMAP.md +++ b/agent-roadmap/ROADMAP.md @@ -29,7 +29,7 @@ NomadCode는 Flutter 기반 앱, core 서비스, 공유 계약, agent-operation - 요약: Flutter 앱에 섞인 Mattermost push notification migration을 분리했고, 현재 package 경로는 `../nexo/packages/messaging_flutter`다. - [진행중] Workflow Core - 경로: `agent-roadmap/phase/workflow-core/PHASE.md` - - 요약: Roadmap Driven Agent-Ops Automation을 NomadCode Core와 MCP-first 제어 표면 방향으로 정리한 뒤 실제 e2e 흐름을 기준으로 task lifecycle, retry, timeout, notification event를 안정화한다. + - 요약: Roadmap Driven Agent-Ops Automation과 proto-socket 내부 통신 레일을 정리한 뒤 client-core 통신과 실제 e2e 흐름을 기준으로 task lifecycle, retry, timeout, notification event를 안정화한다. - [진행중] External Integration - 경로: `agent-roadmap/phase/external-integration/PHASE.md` - 요약: Workflow Core 재개 전에 Mattermost/Nexo messaging 정합성을 먼저 닫고, 이후 Plane 확장, Mattermost, Agent Integrator, IOP OpenAI API Responses-compatible 호출을 실제 통합 흐름으로 확장한다. diff --git a/agent-roadmap/current.md b/agent-roadmap/current.md index 00bc36e..ec8a692 100644 --- a/agent-roadmap/current.md +++ b/agent-roadmap/current.md @@ -12,6 +12,9 @@ - [진행중] Roadmap Driven Agent-Ops Automation - Phase: `agent-roadmap/phase/workflow-core/PHASE.md` - 경로: `agent-roadmap/phase/workflow-core/milestones/roadmap-driven-agent-ops-automation.md` +- [진행중] Proto-Socket Infrastructure Communication Rail + - Phase: `agent-roadmap/phase/workflow-core/PHASE.md` + - 경로: `agent-roadmap/phase/workflow-core/milestones/proto-socket-infrastructure-communication-rail.md` - [계획] Workflow Core - Phase: `agent-roadmap/phase/workflow-core/PHASE.md` - 경로: `agent-roadmap/phase/workflow-core/milestones/workflow-core.md` diff --git a/agent-roadmap/phase/workflow-core/PHASE.md b/agent-roadmap/phase/workflow-core/PHASE.md index 3c74304..21ae643 100644 --- a/agent-roadmap/phase/workflow-core/PHASE.md +++ b/agent-roadmap/phase/workflow-core/PHASE.md @@ -6,17 +6,21 @@ ## 목표 -Plane 통신 토대와 provider-neutral pipeline 설계 이후, 로드맵 기반 agent-ops 운영 루프를 NomadCode Core와 MCP-first 외부 제어 표면이 품는 방향으로 정리한다. 그 다음 실제 상태 변화를 기준으로 task lifecycle, enqueue/running/completed/failed 흐름을 안정화하고 retry, timeout, notification event의 최소 구조를 추가해 core workflow의 운영 기준을 만든다. +Plane 통신 토대와 provider-neutral pipeline 설계 이후, 로드맵 기반 agent-ops 운영 루프를 NomadCode Core와 MCP-first 외부 제어 표면이 품는 방향으로 정리한다. 내부 런타임 간 신규 통신은 proto-socket 기본 레일로 통일하고, client-core 통신을 먼저 열어 실제 상태 변화를 기준으로 task lifecycle, enqueue/running/completed/failed 흐름을 안정화한다. ## Milestone 흐름 완료된 Milestone은 archive 경로를 가리키고, 검토중, 진행중, 계획, 스케치 또는 보류 Milestone은 이 Phase 하위 `milestones/` 경로를 가리킨다. -완료, 검토중, 진행중, 계획, 스케치 순서를 기본으로 하되, 이번 Phase에서는 `Roadmap Driven Agent-Ops Automation`이 `Workflow Core` 구현의 선행 조건이므로 먼저 둔다. +완료, 검토중, 진행중, 계획, 스케치 순서를 기본으로 하되, 이번 Phase에서는 `Roadmap Driven Agent-Ops Automation`과 `Proto-Socket Infrastructure Communication Rail`이 `Workflow Core` 구현의 선행 조건이므로 먼저 둔다. - [진행중] Roadmap Driven Agent-Ops Automation - 경로: `agent-roadmap/phase/workflow-core/milestones/roadmap-driven-agent-ops-automation.md` - 요약: Workflow Core 구현 전에 사용자가 입력과 출력 검토를 NomadCode 라인에서 해결하고, 외부 에이전트는 MCP를 통해 roadmap/action core를 제어하는 운영 방향을 구현 가능한 계획으로 정리한다. +- [진행중] Proto-Socket Infrastructure Communication Rail + - 경로: `agent-roadmap/phase/workflow-core/milestones/proto-socket-infrastructure-communication-rail.md` + - 요약: 내부 런타임 간 신규 통신을 proto-socket 기본 레일로 통일하고, client-core task 요청과 상태 이벤트가 오갈 수 있는 기반을 만든다. + - [계획] Workflow Core - 경로: `agent-roadmap/phase/workflow-core/milestones/workflow-core.md` - 요약: Roadmap Driven Agent-Ops Automation의 Core/MCP-first 계약을 기준으로 task lifecycle, 상태 전이 책임, 실패 정책, retry/timeout, notification event 모델을 안정화한다. @@ -25,5 +29,6 @@ Plane 통신 토대와 provider-neutral pipeline 설계 이후, 로드맵 기반 - Workflow Core는 canonical task lifecycle, retry/timeout envelope, heartbeat, terminal 상태 기록을 소유한다. - `Roadmap Driven Agent-Ops Automation`이 먼저 상태/책임/자동화 경계를 정리해야 `Workflow Core` 구현의 task lifecycle과 완료 이벤트 계약을 확정할 수 있다. +- 내부 런타임 간 신규 통신은 `proto-socket`을 기본으로 하며, REST/HTTP 예외는 health/readiness, 외부 provider 호환 API, 운영 smoke/curl, public callback, 단순 정적 조회로 제한한다. - 외부 협업 도구 notification 발송, provider별 실제 comment/status 전송, 복잡한 workflow DSL은 후속 Phase로 넘긴다. - IOP/A2A/model runtime 내부 정책은 외부 실행 표면 또는 adapter 책임으로 둔다. diff --git a/agent-roadmap/phase/workflow-core/milestones/proto-socket-infrastructure-communication-rail.md b/agent-roadmap/phase/workflow-core/milestones/proto-socket-infrastructure-communication-rail.md new file mode 100644 index 0000000..3778764 --- /dev/null +++ b/agent-roadmap/phase/workflow-core/milestones/proto-socket-infrastructure-communication-rail.md @@ -0,0 +1,93 @@ +# Milestone: Proto-Socket Infrastructure Communication Rail + +## 위치 + +- Roadmap: `agent-roadmap/ROADMAP.md` +- Phase: `agent-roadmap/phase/workflow-core/PHASE.md` + +## 목표 + +NomadCode 내부 런타임 간 신규 통신을 `proto-socket` 중심으로 통일하는 기반을 만든다. 모든 내부 인프라 통신 후보를 대상으로 REST/HTTP 예외를 분리하고, 첫 구현 축은 Flutter client와 Core가 같은 레일로 task 조회, 실행 요청, 상태 이벤트를 주고받을 수 있게 한다. + +## 상태 + +[진행중] + +## 승격 조건 + +- 없음 + +## 구현 잠금 + +- 상태: 해제 +- 결정 필요: 없음 + +## 범위 + +- 내부 런타임 간 통신 표준을 `proto-socket` 기본, REST/HTTP 예외 방식으로 정리 +- `apps/client`와 `services/core` 사이의 client-core 통신 레일 구현 준비 +- `packages/contracts`에 proto-socket channel, message envelope, error/heartbeat/auth 후보를 정리 +- 기존 core REST task API를 proto-socket channel 후보로 매핑하고 전환 경계를 기록 +- Core의 proto-socket endpoint와 Flutter client connector의 최소 구현 단위 정의 +- health/readiness, 외부 provider 호환 API, 운영 smoke/curl, public callback, 단순 정적 조회 같은 REST/HTTP 예외 목록 관리 + +## 기능 + +### Epic: [protocol] Internal communication contract + +내부 인프라 통신을 proto-socket 중심으로 통일하기 위한 계약과 예외 경계를 정리한다. + +- [x] [standard-boundary] 내부 런타임 간 신규 통신은 `proto-socket`을 기본으로 삼고 REST/HTTP 예외 조건을 contracts 또는 domain rule에서 추적한다. +- [x] [message-envelope] request, response, event, error, heartbeat, auth metadata를 포함한 최소 message envelope 후보를 정리한다. +- [x] [rest-compat-map] 현재 core REST task API와 health/readiness endpoint를 proto-socket channel 후보와 REST 예외 목록으로 매핑한다. + +### Epic: [core-rail] Core proto-socket rail + +Core가 내부 런타임 통신용 proto-socket endpoint와 task channel을 제공할 수 있게 한다. + +- [ ] [core-endpoint] `services/core`에 proto-socket endpoint 구성 위치와 lifecycle을 추가한다. 검증: Core가 기존 REST health/readiness를 유지하면서 proto-socket 연결 endpoint를 노출한다. +- [ ] [task-channels] task create/list/get/enqueue와 task status event를 proto-socket channel로 제공한다. 검증: REST task API와 같은 의미의 요청/응답이 proto-socket 경로에서 동작한다. +- [ ] [core-contract-tests] core proto-socket channel, auth, error, heartbeat 동작을 테스트로 고정한다. + +### Epic: [client-rail] Flutter client-core rail + +Flutter client가 no-op proto-socket facade를 실제 Core 연결 레일로 전환한다. + +- [ ] [client-connector] `apps/client`의 `NoopProtoSocketConnector`를 대체할 concrete connector를 추가한다. 검증: 설정된 endpoint로 연결 상태가 `connected`까지 전이된다. +- [ ] [client-task-api] workspace/task 화면이 mock 데이터 대신 Core proto-socket task channel을 소비할 수 있는 client service 경계를 만든다. +- [ ] [client-tests] proto-socket endpoint 설정, reconnect/failure lifecycle, client task service를 Flutter test로 검증한다. + +### Epic: [ops] Observability and rollout + +내부 통신 레일을 운영 가능하게 만들고 예외 프로토콜이 다시 기본값으로 새지 않게 한다. + +- [ ] [diagnostics] connection id, protocol version, channel name, error code를 로그와 디버그 표면에서 확인할 수 있게 한다. +- [ ] [rollout-docs] 새 내부 통신을 추가할 때 proto-socket 우선 여부와 REST/HTTP 예외 사유를 확인하는 운영 체크를 문서화한다. + +## 완료 리뷰 + +- 상태: 없음 +- 요청일: 없음 +- 완료 근거: 없음 +- 리뷰 필요: + - [ ] 사용자가 완료 결과를 확인했다 + - [ ] archive 이동을 승인했다 +- 리뷰 코멘트: 없음 + +## 범위 제외 + +- Mattermost, Plane, Jira 같은 외부 provider의 native REST API 자체를 proto-socket으로 대체하지 않는다. +- health/readiness, 운영 smoke/curl, public callback, 단순 정적 조회 REST 예외를 제거하지 않는다. +- `../nexo/packages/messaging_flutter`의 native push 처리 로직을 수정하지 않는다. +- IOP native protocol, A2A JSON-RPC, OpenAI-compatible Responses 같은 외부 실행 표면을 이 Milestone에서 재설계하지 않는다. +- 전체 프로젝트 관리 UX 완성은 후속 Project Workspace Management UX 범위로 둔다. + +## 작업 컨텍스트 + +- 관련 경로: `services/core/internal/http/`, `services/core/internal/workflow/`, `services/core/internal/scheduler/`, `apps/client/lib/src/integrations/proto_socket/`, `apps/client/lib/src/features/workspaces/`, `packages/contracts/` +- 표준선(선택): 내부 런타임 간 신규 통신은 `proto-socket`을 기본 표준으로 삼고, REST/HTTP는 명시된 예외 조건에서만 유지한다. +- 선행 작업: Mattermost Nexo Messaging Alignment, Client Integration Standardization, 프로젝트 룰의 proto-socket 표준 규약 +- 후속 작업: Workflow Core, External Integration, Project Workspace Management UX +- 현재 지점: Nexo/Mattermost host 경계는 최신화와 테스트가 끝났고, 다음 작업은 client-core 통신이 가능하도록 내부 통신 레일을 까는 것이다. +- 반영 근거: `packages/contracts/notes/flutter-core-api-candidates.md`에 내부 통신 원칙, proto-socket semantic envelope 후보, REST compatibility map 후보를 추가했다. +- 확인 필요: 없음 diff --git a/agent-roadmap/phase/workflow-core/milestones/workflow-core.md b/agent-roadmap/phase/workflow-core/milestones/workflow-core.md index 03d7da9..2e3dcb1 100644 --- a/agent-roadmap/phase/workflow-core/milestones/workflow-core.md +++ b/agent-roadmap/phase/workflow-core/milestones/workflow-core.md @@ -26,6 +26,7 @@ Plane 통신 토대와 별도 pipeline 설계가 정리된 뒤 실제 상태 변 - notification event 구조 정리 - Plane 통신 토대와 후속 pipeline 설계에서 확인된 상태 변화와 실패 케이스 반영 - Roadmap Driven Agent-Ops Automation의 Core/MCP-first 계약에서 정리된 로드맵/마일스톤/plan/review/완료 이벤트 책임 경계 반영 +- Proto-Socket Infrastructure Communication Rail에서 정리한 client-core 통신 레일과 내부 통신 예외 경계 반영 ## 기능 @@ -60,8 +61,8 @@ Core task lifecycle과 상태 전이 책임, 실패/재시도/timeout/notificati - 이전 출처: `services/core/README.md`의 `## 단계별 다음 작업` - 주요 작업 영역: `services/core/internal/workflow/`, `services/core/internal/scheduler/`, `services/core/internal/notification/` -- 선행 작업: Plane Communication Foundation, Work Item Provider Pipeline Design, Roadmap Driven Agent-Ops Automation -- 선행 조건: Roadmap Driven Agent-Ops Automation이 `[계획]` 상태에서 Core/MCP-first 로드맵 action 계약을 기준선으로 제공한다. +- 선행 작업: Plane Communication Foundation, Work Item Provider Pipeline Design, Roadmap Driven Agent-Ops Automation, Proto-Socket Infrastructure Communication Rail +- 선행 조건: Roadmap Driven Agent-Ops Automation이 Core/MCP-first 로드맵 action 계약을 기준선으로 제공하고, Proto-Socket Infrastructure Communication Rail이 client-core task 요청과 상태 이벤트가 오갈 수 있는 내부 통신 기준선을 제공한다. - 후속 작업: External Integration - 현재 지점: Flutter-first Client Consolidation, Client Integration Standardization, Work Item Provider Pipeline Design이 완료되었고, workflow 상태 전이는 Roadmap Driven Agent-Ops Automation의 Core/MCP-first 방향성을 반영해 진행한다. - 책임 경계: diff --git a/agent-task/archive/2026/05/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/USER_REVIEW.md b/agent-task/archive/2026/05/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/USER_REVIEW.md new file mode 100644 index 0000000..7bce65d --- /dev/null +++ b/agent-task/archive/2026/05/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/USER_REVIEW.md @@ -0,0 +1,80 @@ +# User Review Required - m-proto-socket-infrastructure-communication-rail/01_core_endpoint + +## 요청 일시 + +2026-05-30 + +## 상태 + +RESOLVED + +## 사유 + +- 유형: implementation-blocked +- 현재 리뷰 회차: 3 +- 최종 판정: FAIL +- 요약: proto-socket 원격 module path mismatch 때문에 plan의 필수 조건인 repo-external `replace` 제거를 저장소 안에서 안전하게 완료할 수 없다. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/plan_cloud_G08_0.log` | `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/code_review_cloud_G08_0.log` | FAIL | repo-external `replace`, `retryable` 누락, authenticated route test 누락, gofmt 문제 | +| `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/plan_cloud_G08_1.log` | `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/code_review_cloud_G08_1.log` | FAIL | follow-up implementation이 placeholder 상태였고 이전 Required가 남음 | +| `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/plan_cloud_G08_2.log` | `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/code_review_cloud_G08_2.log` | FAIL | proto-socket module mismatch로 `replace` 제거 불가, 사용자 결정 필요 | + +## 차단 근거 + +- 문제: `services/core/go.mod`가 아직 `git.toki-labs.com/toki/proto-socket/go v0.0.0-00010101000000-000000000000`와 `replace git.toki-labs.com/toki/proto-socket/go => ../../../proto-socket/go`에 의존한다. +- 현재 archive plan: `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/plan_cloud_G08_2.log` +- 현재 archive review: `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/code_review_cloud_G08_2.log` +- 검증 명령: `cd services/core && go mod tidy` +- 실제 출력: + +```text +go: downloading git.toki-labs.com/toki/common-proto-socket/go v0.0.0-20260529210328-6f09e2d721c8 +go: found git.toki-labs.com/toki/common-proto-socket/go/packets in git.toki-labs.com/toki/common-proto-socket/go v0.0.0-20260529210328-6f09e2d721c8 +go: github.com/nomadcode/nomadcode-core/internal/protosocket imports + git.toki-labs.com/toki/proto-socket/go imports + git.toki-labs.com/toki/common-proto-socket/go/packets: git.toki-labs.com/toki/common-proto-socket/go@v0.0.0-20260529210328-6f09e2d721c8: parsing go.mod: + module declares its path as: git.toki-labs.com/toki/proto-socket/go + but was required as: git.toki-labs.com/toki/common-proto-socket/go +``` + +- 차단 판단 근거: local `replace`를 커밋할지, 원격 proto-socket repository/module path를 먼저 고칠지, 또는 dependency 전략을 바꿀지는 repo 내부 구현자가 안전하게 단정할 수 없는 배포/의존성 정책 결정이다. + +## 사용자 결정 필요 + +- [x] 자동 follow-up plan/review를 계속 진행한다. +- [ ] 계획을 재작성한다. +- [ ] 테스트 환경, secret, 외부 서비스, SDK, 장비 조건을 준비한 뒤 재시도한다. +- [ ] 작업 범위를 줄이거나 보류/폐기한다. + +## 해소 내용 + +- 사용자 결정: 원격 `proto-socket` 경로를 현재 환경과 맞춰 사용해도 된다는 승인. +- 적용 결과: `services/core/go.mod`에서 repo-external `replace`를 제거하고, 원격 `git.toki-labs.com/toki/proto-socket/go v0.0.0-20260530100557-fed0da37fa1a`로 고정했다. +- 해소 검증: + +```text +$ GOPRIVATE=git.toki-labs.com/toki go mod tidy +(출력 없음) + +$ GOPRIVATE=git.toki-labs.com/toki go list -m git.toki-labs.com/toki/proto-socket/go +git.toki-labs.com/toki/proto-socket/go v0.0.0-20260530100557-fed0da37fa1a +``` + +## 재개 조건 + +- `services/core/go.mod`에 repo-external `replace`를 커밋해도 되는지 승인한다. +- 또는 `git.toki-labs.com/toki/proto-socket/go` 원격 repository의 module path와 published version을 정합하게 고친 뒤, `go mod tidy`가 `replace` 없이 성공하는 상태를 제공한다. +- 또는 proto-socket dependency를 vendoring/submodule/workspace 등 다른 방식으로 고정하는 새 범위를 승인한다. + +## 다음 실행 힌트 + +- 사용자 결정 후 이 task path에서 plan 루프를 재개한다: `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint`. + +## 종료 규칙 + +- 사용자가 이 stop state를 완료/PASS로 해소하면 `USER_REVIEW.md`를 해소 상태로 갱신하고, `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다. +- 새 구현이 필요하면 `plan` 스킬이 `USER_REVIEW.md`를 `user_review_N.log`로 아카이브한 뒤 새 `PLAN-*-G??.md` / `CODE_REVIEW-*-G??.md`를 작성한다. diff --git a/apps/client/pubspec.lock b/apps/client/pubspec.lock index edab04f..b11ba72 100644 --- a/apps/client/pubspec.lock +++ b/apps/client/pubspec.lock @@ -250,10 +250,10 @@ packages: dependency: transitive description: name: protobuf - sha256: "68645b24e0716782e58948f8467fd42a880f255096a821f9e7d0ec625b00c84d" + sha256: "75ec242d22e950bdcc79ee38dd520ce4ee0bc491d7fadc4ea47694604d22bf06" url: "https://pub.dev" source: hosted - version: "3.1.0" + version: "6.0.0" sky_engine: dependency: transitive description: flutter diff --git a/packages/contracts/notes/flutter-core-api-candidates.md b/packages/contracts/notes/flutter-core-api-candidates.md index 3ed4050..281a03e 100644 --- a/packages/contracts/notes/flutter-core-api-candidates.md +++ b/packages/contracts/notes/flutter-core-api-candidates.md @@ -9,9 +9,122 @@ --- -## 1. 공통 및 인프라 API +## 1. 내부 통신 원칙 -### 1.1 `GET /healthz` +### 1.1 기본 표준 +- **설명**: NomadCode 내부 런타임 간 신규 통신은 `proto-socket`을 기본 표준으로 사용한다. +- **안정성 수준**: 후보 (Candidate) +- **적용 대상**: + - `apps/client` ↔ `services/core` + - NomadCode Core ↔ 내부 control/runtime adapter + - 향후 NomadCode가 직접 소유하는 MCP/action runtime 표면 +- **예외 대상**: + - health/readiness처럼 외부 운영 도구와 curl smoke가 직접 확인해야 하는 endpoint + - Plane/Jira/Mattermost/IOP/OpenAI-compatible Responses/A2A처럼 외부 provider 또는 외부 실행 표면이 요구하는 native protocol + - public callback/webhook + - 단순 정적 조회처럼 `proto-socket` 연결 비용이 과한 read-only endpoint +- **예외 기록 규칙**: REST/HTTP를 새로 추가하거나 유지할 때는 이 문서 또는 관련 domain rule에 예외 사유를 남긴다. + +### 1.2 proto-socket Semantic Envelope 후보 +- **설명**: 아래 JSON은 wire schema가 아니라 proto-socket message가 가져야 할 의미 필드 후보다. 실제 source schema는 proto-socket 계약이 안정화되면 별도 schema로 승격한다. +- **Envelope Shape 후보**: + ```json + { + "protocol_version": "nomadcode.proto-socket.v1", + "id": "msg-uuid", + "correlation_id": "request-or-flow-id", + "type": "request", + "channel": "task", + "action": "task.list", + "actor": { + "kind": "client", + "id": "apps/client" + }, + "auth": { + "scheme": "basic", + "token_ref": "runtime-secret-ref" + }, + "payload": {}, + "error": null, + "meta": { + "connection_id": "conn-id", + "timestamp": "2026-05-30T00:00:00Z" + } + } + ``` +- **type 후보**: + - `request`: 요청 메시지 + - `response`: 요청 성공 응답 + - `event`: 구독 또는 서버 발행 이벤트 + - `error`: 요청 실패 또는 channel-level 오류 + - `heartbeat`: 연결 생존 확인 + - `auth`: 연결 인증 또는 인증 갱신 +- **error 후보 필드**: + ```json + { + "code": "task.not_found", + "message": "task not found", + "retryable": false, + "details": {} + } + ``` + +### 1.3 REST Compatibility Map 후보 +- **설명**: 현재 REST API는 proto-socket rail 구현 전까지의 호환 표면이다. task 관련 신규 내부 기능은 proto-socket channel을 우선한다. + +| 현재 REST/HTTP 표면 | proto-socket 후보 | 유지/전환 기준 | +|--------------------|-------------------|----------------| +| `GET /healthz` | 없음 | Public 운영 health check로 REST 유지 | +| `GET /readyz` | `system.ready` event 또는 request 후보 | DB readiness와 curl smoke 때문에 REST 유지, 내부 UI에는 proto-socket 상태 이벤트를 추가 가능 | +| `POST /api/tasks` | `task.create` request | client-core 내부 호출은 proto-socket으로 전환, REST는 smoke/compat 유지 | +| `GET /api/tasks` | `task.list` request | client-core 내부 조회는 proto-socket으로 전환, REST는 smoke/compat 유지 | +| `GET /api/tasks/{id}` | `task.get` request | client-core 내부 조회는 proto-socket으로 전환, REST는 smoke/compat 유지 | +| `POST /api/tasks/{id}/enqueue` | `task.enqueue` request + `task.status.changed` event | client-core 실행 요청과 상태 반영은 proto-socket으로 전환, REST는 smoke/compat 유지 | +| `POST /api/integrations/plane/tasks` | `workitem.plane.import` 후보 | Plane API 자체는 외부 provider REST이며, 내부 호출 표면은 proto-socket으로 추가 가능 | + +### 1.4 Core Endpoint 후보 +- **기본 경로**: `/proto-socket` +- **인증**: 기존 Core Basic Auth middleware 적용 +- **Wire message**: 초기 구현은 `google.protobuf.Struct` semantic envelope를 사용한다. +- **Heartbeat**: proto-socket 기본 heartbeat를 사용하며 Core 설정 기본값은 interval 30s, wait 10s다. + +### 1.5 Task Channel Actions (구현됨) +- **설명**: `services/core/internal/protosocket`가 REST task API와 같은 의미를 `task` channel의 proto-socket action으로 제공한다. REST는 smoke/compat로 유지한다. +- **안정성 수준**: 후보 (Candidate) +- **채널**: `task` +- **요청/응답 action 표**: + +| action | request payload | response type | response payload | +|--------|-----------------|---------------|------------------| +| `task.create` | `POST /api/tasks` JSON shape (`title`, `source`, `payload`, `metadata?`, `external?`) | `response` | `{ "id", "status", "external_provider", "external_id", "task": { …full task } }` | +| `task.list` | `{ "limit": 20 }` (생략 시 0 전달, 기본 20/최대 100 bounding은 core workflow service가 담당) | `response` | `{ "tasks": [ { …full task } ] }` | +| `task.get` | `{ "id": "task-id" }` | `response` | `{ "task": { …full task } }` | +| `task.enqueue` | `{ "id": "task-id" }` | `response` | `{ "id", "status": "queued", "task": { …full task } }` | + +- **task status event (서버 발행)**: + +| action | type | channel | payload | +|--------|------|---------|---------| +| `task.status.changed` | `event` | `task` | `{ "id", "type": "task.running\|task.completed\|task.failed\|task.canceled", "status", "title", "message" }` | + + - scheduler가 내는 모든 task lifecycle event(running/completed/failed/canceled)가 notification fanout을 통해 broadcast된다. completed event의 Mattermost 알림 동작은 그대로 유지된다. + +- **error envelope code 표** (`type: "error"`): + +| code | 의미 | retryable | REST 대응 status | +|------|------|-----------|------------------| +| `task.invalid_input` | 잘못된 task 입력(`ErrInvalidTaskInput`) | false | 400 | +| `task.conflict` | 현재 상태에서 enqueue 불가(`ErrTaskCannotBeEnqueued`) | false | 409 | +| `task.not_found` | task 없음(`pgx.ErrNoRows`) | false | 404 | +| `task.invalid_payload` | envelope payload 디코드 실패 | false | (REST 없음) | +| `internal.error` | 그 외 내부 오류 | true | 500 | +| `UNSUPPORTED_ACTION` | dispatcher에 등록되지 않은 action | false | (REST 없음) | + +--- + +## 2. 공통 및 인프라 API + +### 2.1 `GET /healthz` - **설명**: 서비스 헬스 체크. - **인증**: 없음 (Public) - **안정성 수준**: 안정 (Stable) @@ -22,7 +135,7 @@ } ``` -### 1.2 `GET /readyz` +### 2.2 `GET /readyz` - **설명**: 서비스 및 데이터베이스 준비 상태 체크. - **인증**: Basic Auth - **안정성 수준**: 안정 (Stable) @@ -41,9 +154,9 @@ --- -## 2. 태스크 관리 API (Tasks API) +## 3. 태스크 관리 API (Tasks API) -### 2.1 `POST /api/tasks` +### 3.1 `POST /api/tasks` - **설명**: 새로운 태스크 생성. - **인증**: Basic Auth - **안정성 수준**: 안정 (Stable) @@ -72,7 +185,7 @@ } ``` -### 2.2 `GET /api/tasks` +### 3.2 `GET /api/tasks` - **설명**: 태스크 목록 조회. - **인증**: Basic Auth - **안정성 수준**: 안정 (Stable) @@ -100,7 +213,7 @@ ] ``` -### 2.3 `GET /api/tasks/{id}` +### 3.3 `GET /api/tasks/{id}` - **설명**: 단일 태스크 세부 정보 조회. - **인증**: Basic Auth - **안정성 수준**: 안정 (Stable) @@ -124,7 +237,7 @@ } ``` -### 2.4 `POST /api/tasks/{id}/enqueue` +### 3.4 `POST /api/tasks/{id}/enqueue` - **설명**: 태스크를 큐에 진입시켜 실행 대기 상태로 변경. - **인증**: Basic Auth - **안정성 수준**: 안정 (Stable) @@ -138,9 +251,9 @@ --- -## 3. 외부 연동 API (Integrations API) +## 4. 외부 연동 API (Integrations API) -### 3.1 `POST /api/integrations/plane/tasks` +### 4.1 `POST /api/integrations/plane/tasks` - **설명**: Plane 워크아이템 정보를 기반으로 새로운 태스크 생성. - **인증**: Basic Auth - **안정성 수준**: 안정 (Stable) @@ -167,12 +280,12 @@ --- -## 4. Workspace / Project Metadata 계약 후보 +## 5. Workspace / Project Metadata 계약 후보 > [!IMPORTANT] > **Workspace/Project Metadata 상태**: 현재 Go core 백엔드에 이와 관련한 별도의 Source-of-Truth 저장소나 전용 테이블 구조가 존재하지 않으며, API 명세 또한 완전히 확정되지 않았습니다. 아래는 Flutter 앱과 코어 간 향후 연동을 대비한 **계약 후보 필드 목록**입니다. -### 4.1 Workspace Metadata 후보 스펙 +### 5.1 Workspace Metadata 후보 스펙 - **설명**: 에이전트 작업 공간(Workspace)에 대한 설정 및 상태 메타데이터. - **후보 필드**: - `workspace_id`: 작업 공간 고유 식별자 (UUID) @@ -184,7 +297,7 @@ - `status`: 현재 상태 (예: `active`, `archived`, `suspended`) - `settings`: 작업 공간에 특화된 동적 설정 JSON 오브젝트 (예: 자동 스케줄링 옵션, 알림 채널 정보 등) -### 4.2 Project Metadata 후보 스펙 +### 5.2 Project Metadata 후보 스펙 - **설명**: 작업 공간 내 세부 프로젝트(Project) 정보. - **후보 필드**: - `project_id`: 프로젝트 고유 식별자 (UUID) @@ -197,13 +310,13 @@ --- -## 5. 클라이언트 Integration 설정 계약 후보 +## 6. 클라이언트 Integration 설정 계약 후보 > [!IMPORTANT] > 이 절은 `apps/client`의 integration boundary에서 host/plugin/transport 사이에 교환되는 설정값을 후보로 모은 것이다. 실제 source schema는 소비하는 core/client 경계가 안정화된 뒤로 미룬다. > 관련 코드 경계: `apps/client/lib/src/integrations/proto_socket/` 와 `apps/client/lib/src/integrations/mattermost/`. -### 5.1 proto-socket Endpoint 설정 후보 +### 6.1 proto-socket Endpoint 설정 후보 - **설명**: `apps/client`가 proto-socket transport에 의존할 때 사용하는 연결/하트비트 설정. 현재 구현은 `proto_socket` Dart 패키지(local path)를 끌어 쓰며 bootstrap에서 자동 연결하지 않는다. - **후보 필드**: - `host`: 대상 서버 hostname @@ -214,7 +327,7 @@ - `heartbeat_wait_seconds`: heartbeat 응답 대기 한계 (기본 60) - `enabled`: bootstrap 시 자동 연결 활성화 여부 (기본 false) -### 5.2 Mattermost Push Host 책임 경계 후보 +### 6.2 Mattermost Push Host 책임 경계 후보 - **설명**: Mattermost push 통합에서 `apps/client` host가 plugin adapter에게 위임/공급해야 하는 데이터와 콜백. plugin adapter는 platform 측 plugin singleton에 격리되어 있고, 그 외 코드는 모두 `MattermostPushClient` 인터페이스에만 의존한다. - **호스트 소유 필드/콜백**: - `server_url`: Mattermost 서버 base URL (auth 핸드오프, signing key 저장에 함께 사용) diff --git a/services/core/cmd/server/main.go b/services/core/cmd/server/main.go index 3fe9a64..572209b 100644 --- a/services/core/cmd/server/main.go +++ b/services/core/cmd/server/main.go @@ -19,6 +19,7 @@ import ( "github.com/nomadcode/nomadcode-core/internal/db" apphttp "github.com/nomadcode/nomadcode-core/internal/http" "github.com/nomadcode/nomadcode-core/internal/notification" + "github.com/nomadcode/nomadcode-core/internal/protosocket" "github.com/nomadcode/nomadcode-core/internal/scheduler" "github.com/nomadcode/nomadcode-core/internal/storage" "github.com/nomadcode/nomadcode-core/internal/workflow" @@ -53,7 +54,12 @@ func run(logger *slog.Logger) error { BaseURL: cfg.PlaneBaseURL, Token: cfg.PlaneToken, }, logger) - notificationService := notification.NewService(mattermostClient, logger) + protoSocketServer := protosocket.NewServer(protosocket.Config{ + HeartbeatIntervalSec: cfg.ProtoSocketHeartbeatIntervalSec, + HeartbeatWaitSec: cfg.ProtoSocketHeartbeatWaitSec, + }, logger) + + notificationService := notification.NewService(mattermostClient, logger, protosocket.NewTaskEventBroadcaster(protoSocketServer)) modelClient := modelopenai.NewClient(modelopenai.Config{ BaseURL: cfg.ModelBaseURL, APIKey: cfg.ModelAPIKey, @@ -87,9 +93,12 @@ func run(logger *slog.Logger) error { workflowService := workflow.NewService(store, taskScheduler, logger) handler := apphttp.NewHandler(pool, workflowService, planeClient, logger) + + protosocket.NewTaskChannels(workflowService).Register(protoSocketServer.Dispatcher()) + server := &http.Server{ Addr: cfg.HTTPAddr, - Handler: apphttp.NewRouter(handler, logger, apphttp.AuthConfig{Username: cfg.AuthUsername, Password: cfg.AuthPassword}), + Handler: apphttp.NewRouter(handler, logger, apphttp.AuthConfig{Username: cfg.AuthUsername, Password: cfg.AuthPassword}, protoSocketServer, cfg.ProtoSocketPath), ReadHeaderTimeout: 5 * time.Second, } @@ -117,6 +126,9 @@ func run(logger *slog.Logger) error { if err := server.Shutdown(shutdownCtx); err != nil { return err } + if err := protoSocketServer.Close(); err != nil { + return err + } if err := taskScheduler.Stop(shutdownCtx); err != nil { return err } diff --git a/services/core/go.mod b/services/core/go.mod index 20f7431..283c5c9 100644 --- a/services/core/go.mod +++ b/services/core/go.mod @@ -3,10 +3,13 @@ module github.com/nomadcode/nomadcode-core go 1.25.0 require ( + git.toki-labs.com/toki/proto-socket/go v0.0.0-20260530100557-fed0da37fa1a github.com/go-chi/chi/v5 v5.2.5 github.com/jackc/pgx/v5 v5.9.2 github.com/riverqueue/river v0.37.1 github.com/riverqueue/river/riverdriver/riverpgxv5 v0.37.1 + google.golang.org/protobuf v1.36.11 + nhooyr.io/websocket v1.8.17 ) require ( diff --git a/services/core/go.sum b/services/core/go.sum index 9d7375f..abfae34 100644 --- a/services/core/go.sum +++ b/services/core/go.sum @@ -1,8 +1,12 @@ +git.toki-labs.com/toki/proto-socket/go v0.0.0-20260530100557-fed0da37fa1a h1:o9SRJC3AYzBuO+/WVsXJnW+OQFj0sySoKhWRI2bnMYM= +git.toki-labs.com/toki/proto-socket/go v0.0.0-20260530100557-fed0da37fa1a/go.mod h1:ZQOfKXqiRUwRVWwe3tVVFSOKka31Wya8vzUfYChDo60= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/go-chi/chi/v5 v5.2.5 h1:Eg4myHZBjyvJmAFjFvWgrqDTXFyOzjj7YIm3L3mu6Ug= github.com/go-chi/chi/v5 v5.2.5/go.mod h1:X7Gx4mteadT3eDOMTsXzmI4/rwUpOwBHLpAfupzFJP0= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438 h1:Dj0L5fhJ9F82ZJyVOmBx6msDp/kfd1t9GRfny/mfJA0= github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= @@ -55,9 +59,13 @@ golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +nhooyr.io/websocket v1.8.17 h1:KEVeLJkUywCKVsnLIDlD/5gtayKp8VoCkksHCGGfT9Y= +nhooyr.io/websocket v1.8.17/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c= diff --git a/services/core/internal/config/config.go b/services/core/internal/config/config.go index 39d92ed..316eaad 100644 --- a/services/core/internal/config/config.go +++ b/services/core/internal/config/config.go @@ -6,52 +6,58 @@ import ( ) type Config struct { - AppEnv string - HTTPAddr string - DatabaseURL string - RedisURL string - RedisKeyPrefix string - AuthUsername string - AuthPassword string - ModelBaseURL string - ModelAPIKey string - ModelName string - ModelContextSize int - ModelTimeoutSec int - A2AEdgeURL string - A2AAgentURL string - A2AToken string - A2ATimeoutSec int - MattermostBaseURL string - MattermostToken string - PlaneBaseURL string - PlaneToken string - WorkflowTaskTimeoutSec int + AppEnv string + HTTPAddr string + DatabaseURL string + RedisURL string + RedisKeyPrefix string + AuthUsername string + AuthPassword string + ModelBaseURL string + ModelAPIKey string + ModelName string + ModelContextSize int + ModelTimeoutSec int + A2AEdgeURL string + A2AAgentURL string + A2AToken string + A2ATimeoutSec int + MattermostBaseURL string + MattermostToken string + PlaneBaseURL string + PlaneToken string + WorkflowTaskTimeoutSec int + ProtoSocketPath string + ProtoSocketHeartbeatIntervalSec int + ProtoSocketHeartbeatWaitSec int } func Load() Config { return Config{ - AppEnv: getEnv("APP_ENV", "local"), - HTTPAddr: getEnv("HTTP_ADDR", ":8080"), - DatabaseURL: os.Getenv("DATABASE_URL"), - RedisURL: os.Getenv("REDIS_URL"), - RedisKeyPrefix: os.Getenv("REDIS_KEY_PREFIX"), - AuthUsername: getEnv("AUTH_USERNAME", "nomadcode"), - AuthPassword: os.Getenv("AUTH_PASSWORD"), - ModelBaseURL: getEnv("MODEL_BASE_URL", "http://192.168.0.91:11434"), - ModelAPIKey: getEnv("MODEL_API_KEY", "ollama"), - ModelName: getEnv("MODEL_NAME", "qwen3.6:35b-a3b-bf16"), - ModelContextSize: getEnvInt("MODEL_CONTEXT_SIZE", 262144), - ModelTimeoutSec: getEnvInt("MODEL_TIMEOUT_SEC", 300), - A2AEdgeURL: firstEnv("A2A_EDGE_URL", "A2A_AGENT_URL"), - A2AAgentURL: firstEnv("A2A_AGENT_URL", "A2A_EDGE_URL"), - A2AToken: os.Getenv("A2A_TOKEN"), - A2ATimeoutSec: getEnvInt("A2A_TIMEOUT_SEC", 300), - MattermostBaseURL: os.Getenv("MATTERMOST_BASE_URL"), - MattermostToken: os.Getenv("MATTERMOST_TOKEN"), - PlaneBaseURL: os.Getenv("PLANE_BASE_URL"), - PlaneToken: os.Getenv("PLANE_TOKEN"), - WorkflowTaskTimeoutSec: getEnvInt("WORKFLOW_TASK_TIMEOUT_SEC", 300), + AppEnv: getEnv("APP_ENV", "local"), + HTTPAddr: getEnv("HTTP_ADDR", ":8080"), + DatabaseURL: os.Getenv("DATABASE_URL"), + RedisURL: os.Getenv("REDIS_URL"), + RedisKeyPrefix: os.Getenv("REDIS_KEY_PREFIX"), + AuthUsername: getEnv("AUTH_USERNAME", "nomadcode"), + AuthPassword: os.Getenv("AUTH_PASSWORD"), + ModelBaseURL: getEnv("MODEL_BASE_URL", "http://192.168.0.91:11434"), + ModelAPIKey: getEnv("MODEL_API_KEY", "ollama"), + ModelName: getEnv("MODEL_NAME", "qwen3.6:35b-a3b-bf16"), + ModelContextSize: getEnvInt("MODEL_CONTEXT_SIZE", 262144), + ModelTimeoutSec: getEnvInt("MODEL_TIMEOUT_SEC", 300), + A2AEdgeURL: firstEnv("A2A_EDGE_URL", "A2A_AGENT_URL"), + A2AAgentURL: firstEnv("A2A_AGENT_URL", "A2A_EDGE_URL"), + A2AToken: os.Getenv("A2A_TOKEN"), + A2ATimeoutSec: getEnvInt("A2A_TIMEOUT_SEC", 300), + MattermostBaseURL: os.Getenv("MATTERMOST_BASE_URL"), + MattermostToken: os.Getenv("MATTERMOST_TOKEN"), + PlaneBaseURL: os.Getenv("PLANE_BASE_URL"), + PlaneToken: os.Getenv("PLANE_TOKEN"), + WorkflowTaskTimeoutSec: getEnvInt("WORKFLOW_TASK_TIMEOUT_SEC", 300), + ProtoSocketPath: getEnv("PROTO_SOCKET_PATH", "/proto-socket"), + ProtoSocketHeartbeatIntervalSec: getEnvInt("PROTO_SOCKET_HEARTBEAT_INTERVAL_SEC", 30), + ProtoSocketHeartbeatWaitSec: getEnvInt("PROTO_SOCKET_HEARTBEAT_WAIT_SEC", 10), } } diff --git a/services/core/internal/config/config_test.go b/services/core/internal/config/config_test.go index a7c6c0a..ab2e41e 100644 --- a/services/core/internal/config/config_test.go +++ b/services/core/internal/config/config_test.go @@ -64,3 +64,31 @@ func TestConfigLoadsWorkflowTaskTimeout(t *testing.T) { t.Fatalf("WorkflowTaskTimeoutSec override: got %d, want 120", cfgOverride.WorkflowTaskTimeoutSec) } } + +func TestConfigLoadsProtoSocketDefaultsAndOverrides(t *testing.T) { + cfg := Load() + if cfg.ProtoSocketPath != "/proto-socket" { + t.Fatalf("ProtoSocketPath default: got %q, want %q", cfg.ProtoSocketPath, "/proto-socket") + } + if cfg.ProtoSocketHeartbeatIntervalSec != 30 { + t.Fatalf("ProtoSocketHeartbeatIntervalSec default: got %d, want 30", cfg.ProtoSocketHeartbeatIntervalSec) + } + if cfg.ProtoSocketHeartbeatWaitSec != 10 { + t.Fatalf("ProtoSocketHeartbeatWaitSec default: got %d, want 10", cfg.ProtoSocketHeartbeatWaitSec) + } + + t.Setenv("PROTO_SOCKET_PATH", "/custom-path") + t.Setenv("PROTO_SOCKET_HEARTBEAT_INTERVAL_SEC", "15") + t.Setenv("PROTO_SOCKET_HEARTBEAT_WAIT_SEC", "5") + + cfgOverride := Load() + if cfgOverride.ProtoSocketPath != "/custom-path" { + t.Fatalf("ProtoSocketPath override: got %q, want %q", cfgOverride.ProtoSocketPath, "/custom-path") + } + if cfgOverride.ProtoSocketHeartbeatIntervalSec != 15 { + t.Fatalf("ProtoSocketHeartbeatIntervalSec override: got %d, want 15", cfgOverride.ProtoSocketHeartbeatIntervalSec) + } + if cfgOverride.ProtoSocketHeartbeatWaitSec != 5 { + t.Fatalf("ProtoSocketHeartbeatWaitSec override: got %d, want 5", cfgOverride.ProtoSocketHeartbeatWaitSec) + } +} diff --git a/services/core/internal/http/middleware.go b/services/core/internal/http/middleware.go index fe6b8b1..4e90f6d 100644 --- a/services/core/internal/http/middleware.go +++ b/services/core/internal/http/middleware.go @@ -1,8 +1,11 @@ package http import ( + "bufio" "crypto/subtle" + "errors" "log/slog" + "net" stdhttp "net/http" "time" ) @@ -17,6 +20,16 @@ func (r *statusRecorder) WriteHeader(status int) { r.ResponseWriter.WriteHeader(status) } +// Hijack lets WebSocket upgrades (e.g. the proto-socket route) take over the +// connection even though the response writer is wrapped for status logging. +func (r *statusRecorder) Hijack() (net.Conn, *bufio.ReadWriter, error) { + hijacker, ok := r.ResponseWriter.(stdhttp.Hijacker) + if !ok { + return nil, nil, errors.New("underlying ResponseWriter does not implement http.Hijacker") + } + return hijacker.Hijack() +} + func loggingMiddleware(logger *slog.Logger) func(stdhttp.Handler) stdhttp.Handler { return func(next stdhttp.Handler) stdhttp.Handler { return stdhttp.HandlerFunc(func(w stdhttp.ResponseWriter, r *stdhttp.Request) { diff --git a/services/core/internal/http/middleware_test.go b/services/core/internal/http/middleware_test.go index d0581d6..c954418 100644 --- a/services/core/internal/http/middleware_test.go +++ b/services/core/internal/http/middleware_test.go @@ -44,10 +44,14 @@ func TestBasicAuthMiddlewareAcceptsValidCredentials(t *testing.T) { } func TestRouterKeepsHealthzPublicAndProtectsAPI(t *testing.T) { + protoSocketMock := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + router := NewRouter(NewHandler(nil, nil, nil, nil), nil, AuthConfig{ Username: "nomadcode", Password: "secret", - }) + }, protoSocketMock, "/proto-socket") healthReq := httptest.NewRequest(http.MethodGet, "/healthz", nil) healthRec := httptest.NewRecorder() @@ -64,6 +68,25 @@ func TestRouterKeepsHealthzPublicAndProtectsAPI(t *testing.T) { if apiRec.Code != http.StatusUnauthorized { t.Fatalf("expected /api/tasks status %d, got %d", http.StatusUnauthorized, apiRec.Code) } + + // Verify /proto-socket is protected + wsReq := httptest.NewRequest(http.MethodGet, "/proto-socket", nil) + wsRec := httptest.NewRecorder() + router.ServeHTTP(wsRec, wsReq) + + if wsRec.Code != http.StatusUnauthorized { + t.Fatalf("expected /proto-socket status %d, got %d", http.StatusUnauthorized, wsRec.Code) + } + + // Verify /proto-socket works with valid credentials + wsAuthReq := httptest.NewRequest(http.MethodGet, "/proto-socket", nil) + wsAuthReq.SetBasicAuth("nomadcode", "secret") + wsAuthRec := httptest.NewRecorder() + router.ServeHTTP(wsAuthRec, wsAuthReq) + + if wsAuthRec.Code != http.StatusOK { + t.Fatalf("expected /proto-socket with auth status %d, got %d", http.StatusOK, wsAuthRec.Code) + } } func okHandler() http.Handler { diff --git a/services/core/internal/http/router.go b/services/core/internal/http/router.go index c1c104b..106ff57 100644 --- a/services/core/internal/http/router.go +++ b/services/core/internal/http/router.go @@ -13,7 +13,7 @@ type AuthConfig struct { Password string } -func NewRouter(handler *Handler, logger *slog.Logger, auth AuthConfig) stdhttp.Handler { +func NewRouter(handler *Handler, logger *slog.Logger, auth AuthConfig, protoSocket stdhttp.Handler, protoSocketPath string) stdhttp.Handler { r := chi.NewRouter() r.Use(chimiddleware.RequestID) r.Use(chimiddleware.RealIP) @@ -26,6 +26,10 @@ func NewRouter(handler *Handler, logger *slog.Logger, auth AuthConfig) stdhttp.H protected.Use(basicAuthMiddleware(auth)) protected.Get("/readyz", handler.Readyz) + if protoSocket != nil && protoSocketPath != "" { + protected.Handle(protoSocketPath, protoSocket) + } + protected.Route("/api", func(r chi.Router) { r.Post("/integrations/plane/tasks", handler.CreatePlaneTask) r.Post("/tasks", handler.CreateTask) diff --git a/services/core/internal/notification/model.go b/services/core/internal/notification/model.go index 9a14e7f..3df080c 100644 --- a/services/core/internal/notification/model.go +++ b/services/core/internal/notification/model.go @@ -1,5 +1,13 @@ package notification +import "context" + +// TaskEventSink receives every task event so non-Mattermost consumers (e.g. the +// proto-socket broadcaster) can fan out task status changes. +type TaskEventSink interface { + HandleTaskEvent(context.Context, TaskEvent) error +} + type TaskNotification struct { TaskID string Title string diff --git a/services/core/internal/notification/service.go b/services/core/internal/notification/service.go index 03ab971..a2b7379 100644 --- a/services/core/internal/notification/service.go +++ b/services/core/internal/notification/service.go @@ -2,6 +2,7 @@ package notification import ( "context" + "errors" "log/slog" "github.com/nomadcode/nomadcode-core/internal/adapters/mattermost" @@ -10,12 +11,14 @@ import ( type Service struct { mattermost *mattermost.Client logger *slog.Logger + sinks []TaskEventSink } -func NewService(mattermostClient *mattermost.Client, logger *slog.Logger) *Service { +func NewService(mattermostClient *mattermost.Client, logger *slog.Logger, sinks ...TaskEventSink) *Service { return &Service{ mattermost: mattermostClient, logger: logger, + sinks: append([]TaskEventSink{}, sinks...), } } @@ -34,18 +37,21 @@ func (s *Service) NotifyTaskEvent(ctx context.Context, input TaskEvent) error { s.logger.Info("task event notification requested", "type", input.Type, "task_id", input.TaskID) } - if input.Type != TaskEventCompleted { - return nil + var errs []error + for _, sink := range s.sinks { + if err := sink.HandleTaskEvent(ctx, input); err != nil { + errs = append(errs, err) + } } - if s.mattermost == nil { - return nil + if input.Type == TaskEventCompleted && s.mattermost != nil { + errs = append(errs, s.mattermost.SendTaskNotification(ctx, mattermost.TaskNotificationInput{ + TaskID: input.TaskID, + Title: input.Title, + Status: input.Status, + Message: input.Message, + })) } - return s.mattermost.SendTaskNotification(ctx, mattermost.TaskNotificationInput{ - TaskID: input.TaskID, - Title: input.Title, - Status: input.Status, - Message: input.Message, - }) + return errors.Join(errs...) } diff --git a/services/core/internal/notification/service_test.go b/services/core/internal/notification/service_test.go index 4404320..1e32bf8 100644 --- a/services/core/internal/notification/service_test.go +++ b/services/core/internal/notification/service_test.go @@ -2,12 +2,23 @@ package notification_test import ( "context" + "errors" "log/slog" "testing" "github.com/nomadcode/nomadcode-core/internal/notification" ) +type recordingSink struct { + events []notification.TaskEvent + err error +} + +func (s *recordingSink) HandleTaskEvent(_ context.Context, event notification.TaskEvent) error { + s.events = append(s.events, event) + return s.err +} + type logRecord struct { msg string args map[string]any @@ -79,6 +90,52 @@ func TestNotifyTaskCompletedUsesTaskEventCompleted(t *testing.T) { } } +func TestNotifyTaskEventFansOutToSinksForRunningAndFailed(t *testing.T) { + logger := slog.New(&spyHandler{}) + sink := &recordingSink{} + service := notification.NewService(nil, logger, sink) + + events := []notification.TaskEvent{ + {Type: notification.TaskEventRunning, TaskID: "task-1", Status: "running"}, + {Type: notification.TaskEventFailed, TaskID: "task-2", Status: "failed"}, + {Type: notification.TaskEventCompleted, TaskID: "task-3", Status: "completed"}, + } + for _, e := range events { + if err := service.NotifyTaskEvent(context.Background(), e); err != nil { + t.Fatalf("unexpected error for %s: %v", e.Type, err) + } + } + + if len(sink.events) != 3 { + t.Fatalf("expected sink to receive 3 events, got %d", len(sink.events)) + } + if sink.events[0].Type != notification.TaskEventRunning || sink.events[1].Type != notification.TaskEventFailed { + t.Errorf("sink did not receive non-completed events: %+v", sink.events) + } + if sink.events[2].Type != notification.TaskEventCompleted { + t.Errorf("sink did not receive completed event: %+v", sink.events) + } +} + +func TestNotifyTaskEventReturnsSinkErrors(t *testing.T) { + logger := slog.New(&spyHandler{}) + sinkErr := errors.New("sink boom") + sink := &recordingSink{err: sinkErr} + service := notification.NewService(nil, logger, sink) + + err := service.NotifyTaskEvent(context.Background(), notification.TaskEvent{ + Type: notification.TaskEventRunning, + TaskID: "task-1", + Status: "running", + }) + if err == nil { + t.Fatal("expected sink error to be returned") + } + if !errors.Is(err, sinkErr) { + t.Errorf("expected joined error to wrap sink error, got %v", err) + } +} + func TestNotifyTaskEventAllowsFailedWithoutMattermost(t *testing.T) { spy := &spyHandler{} logger := slog.New(spy) diff --git a/services/core/internal/protosocket/dispatcher.go b/services/core/internal/protosocket/dispatcher.go new file mode 100644 index 0000000..02856c0 --- /dev/null +++ b/services/core/internal/protosocket/dispatcher.go @@ -0,0 +1,59 @@ +package protosocket + +import ( + "context" + "crypto/rand" + "encoding/hex" + "fmt" + "sync" +) + +type HandlerFunc func(context.Context, Envelope) Envelope + +type Dispatcher struct { + mu sync.RWMutex + handlers map[string]HandlerFunc +} + +func NewDispatcher() *Dispatcher { + return &Dispatcher{ + handlers: make(map[string]HandlerFunc), + } +} + +func (d *Dispatcher) Register(action string, handler HandlerFunc) { + d.mu.Lock() + defer d.mu.Unlock() + d.handlers[action] = handler +} + +func (d *Dispatcher) Dispatch(ctx context.Context, env Envelope) Envelope { + d.mu.RLock() + handler, ok := d.handlers[env.Action] + d.mu.RUnlock() + + if !ok { + return Envelope{ + ProtocolVersion: ProtocolVersion, + ID: generateID(), + CorrelationID: env.ID, + Type: "error", + Channel: env.Channel, + Action: env.Action, + Error: &EnvelopeError{ + Code: "UNSUPPORTED_ACTION", + Message: fmt.Sprintf("unsupported action: %s", env.Action), + }, + } + } + + return handler(ctx, env) +} + +func generateID() string { + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + return "error-id-fallback" + } + return hex.EncodeToString(b) +} diff --git a/services/core/internal/protosocket/envelope.go b/services/core/internal/protosocket/envelope.go new file mode 100644 index 0000000..53f16b4 --- /dev/null +++ b/services/core/internal/protosocket/envelope.go @@ -0,0 +1,121 @@ +package protosocket + +import ( + "fmt" + + "google.golang.org/protobuf/types/known/structpb" +) + +const ProtocolVersion = "nomadcode.proto-socket.v1" + +type Envelope struct { + ProtocolVersion string `json:"protocol_version"` + ID string `json:"id"` + CorrelationID string `json:"correlation_id"` + Type string `json:"type"` + Channel string `json:"channel"` + Action string `json:"action"` + Actor map[string]any `json:"actor,omitempty"` + Auth map[string]any `json:"auth,omitempty"` + Payload map[string]any `json:"payload,omitempty"` + Error *EnvelopeError `json:"error,omitempty"` + Meta map[string]any `json:"meta,omitempty"` +} + +type EnvelopeError struct { + Code string `json:"code"` + Message string `json:"message"` + Retryable bool `json:"retryable"` + Details map[string]any `json:"details,omitempty"` +} + +func (e Envelope) ToStruct() (*structpb.Struct, error) { + m := map[string]any{ + "protocol_version": e.ProtocolVersion, + "id": e.ID, + "correlation_id": e.CorrelationID, + "type": e.Type, + "channel": e.Channel, + "action": e.Action, + } + if e.Actor != nil { + m["actor"] = e.Actor + } + if e.Auth != nil { + m["auth"] = e.Auth + } + if e.Payload != nil { + m["payload"] = e.Payload + } + if e.Meta != nil { + m["meta"] = e.Meta + } + if e.Error != nil { + errMap := map[string]any{ + "code": e.Error.Code, + "message": e.Error.Message, + "retryable": e.Error.Retryable, + } + if e.Error.Details != nil { + errMap["details"] = e.Error.Details + } + m["error"] = errMap + } + return structpb.NewStruct(m) +} + +func EnvelopeFromStruct(s *structpb.Struct) (Envelope, error) { + var env Envelope + if s == nil { + return env, fmt.Errorf("nil struct") + } + m := s.AsMap() + + if v, ok := m["protocol_version"].(string); ok { + env.ProtocolVersion = v + } + if v, ok := m["id"].(string); ok { + env.ID = v + } + if v, ok := m["correlation_id"].(string); ok { + env.CorrelationID = v + } + if v, ok := m["type"].(string); ok { + env.Type = v + } + if v, ok := m["channel"].(string); ok { + env.Channel = v + } + if v, ok := m["action"].(string); ok { + env.Action = v + } + if v, ok := m["actor"].(map[string]any); ok { + env.Actor = v + } + if v, ok := m["auth"].(map[string]any); ok { + env.Auth = v + } + if v, ok := m["payload"].(map[string]any); ok { + env.Payload = v + } + if v, ok := m["meta"].(map[string]any); ok { + env.Meta = v + } + if errVal, ok := m["error"].(map[string]any); ok { + var envErr EnvelopeError + if code, ok := errVal["code"].(string); ok { + envErr.Code = code + } + if msg, ok := errVal["message"].(string); ok { + envErr.Message = msg + } + if ret, ok := errVal["retryable"].(bool); ok { + envErr.Retryable = ret + } + if details, ok := errVal["details"].(map[string]any); ok { + envErr.Details = details + } + env.Error = &envErr + } + return env, nil +} diff --git a/services/core/internal/protosocket/envelope_test.go b/services/core/internal/protosocket/envelope_test.go new file mode 100644 index 0000000..6000f03 --- /dev/null +++ b/services/core/internal/protosocket/envelope_test.go @@ -0,0 +1,116 @@ +package protosocket + +import ( + "context" + "testing" +) + +func TestEnvelopeStructRoundTrip(t *testing.T) { + original := Envelope{ + ProtocolVersion: ProtocolVersion, + ID: "msg-123", + CorrelationID: "corr-456", + Type: "request", + Channel: "tasks", + Action: "create", + Actor: map[string]any{"id": "user-1"}, + Auth: map[string]any{"token": "secret"}, + Payload: map[string]any{"title": "hello"}, + Meta: map[string]any{"source": "client"}, + Error: &EnvelopeError{ + Code: "SOME_ERROR", + Message: "An error occurred", + Retryable: true, + Details: map[string]any{"retry": true}, + }, + } + + s, err := original.ToStruct() + if err != nil { + t.Fatalf("failed to convert Envelope to Struct: %v", err) + } + + roundtrip, err := EnvelopeFromStruct(s) + if err != nil { + t.Fatalf("failed to convert Struct back to Envelope: %v", err) + } + + if roundtrip.ProtocolVersion != original.ProtocolVersion { + t.Errorf("ProtocolVersion mismatch: got %q, want %q", roundtrip.ProtocolVersion, original.ProtocolVersion) + } + if roundtrip.ID != original.ID { + t.Errorf("ID mismatch: got %q, want %q", roundtrip.ID, original.ID) + } + if roundtrip.CorrelationID != original.CorrelationID { + t.Errorf("CorrelationID mismatch: got %q, want %q", roundtrip.CorrelationID, original.CorrelationID) + } + if roundtrip.Type != original.Type { + t.Errorf("Type mismatch: got %q, want %q", roundtrip.Type, original.Type) + } + if roundtrip.Channel != original.Channel { + t.Errorf("Channel mismatch: got %q, want %q", roundtrip.Channel, original.Channel) + } + if roundtrip.Action != original.Action { + t.Errorf("Action mismatch: got %q, want %q", roundtrip.Action, original.Action) + } + + if roundtrip.Actor["id"] != "user-1" { + t.Errorf("Actor mismatch: got %v", roundtrip.Actor) + } + if roundtrip.Auth["token"] != "secret" { + t.Errorf("Auth mismatch: got %v", roundtrip.Auth) + } + if roundtrip.Payload["title"] != "hello" { + t.Errorf("Payload mismatch: got %v", roundtrip.Payload) + } + if roundtrip.Meta["source"] != "client" { + t.Errorf("Meta mismatch: got %v", roundtrip.Meta) + } + + if roundtrip.Error == nil { + t.Fatal("Error should not be nil") + } + if roundtrip.Error.Code != "SOME_ERROR" { + t.Errorf("Error.Code mismatch: got %q, want %q", roundtrip.Error.Code, "SOME_ERROR") + } + if roundtrip.Error.Message != "An error occurred" { + t.Errorf("Error.Message mismatch: got %q, want %q", roundtrip.Error.Message, "An error occurred") + } + if !roundtrip.Error.Retryable { + t.Errorf("Error.Retryable mismatch: got false, want true") + } + if roundtrip.Error.Details["retry"] != true { + t.Errorf("Error.Details mismatch: got %v", roundtrip.Error.Details) + } +} + +func TestDispatcherReturnsUnsupportedActionErrorEnvelope(t *testing.T) { + dispatcher := NewDispatcher() + ctx := context.Background() + + request := Envelope{ + ProtocolVersion: ProtocolVersion, + ID: "req-1", + Type: "request", + Channel: "tasks", + Action: "unknown-action", + } + + response := dispatcher.Dispatch(ctx, request) + + if response.Type != "error" { + t.Errorf("expected response type 'error', got %q", response.Type) + } + if response.CorrelationID != "req-1" { + t.Errorf("expected CorrelationID 'req-1', got %q", response.CorrelationID) + } + if response.Error == nil { + t.Fatal("expected non-nil Error in response") + } + if response.Error.Code != "UNSUPPORTED_ACTION" { + t.Errorf("expected error code 'UNSUPPORTED_ACTION', got %q", response.Error.Code) + } + if response.Error.Retryable { + t.Error("expected Retryable to be false for UNSUPPORTED_ACTION") + } +} diff --git a/services/core/internal/protosocket/events.go b/services/core/internal/protosocket/events.go new file mode 100644 index 0000000..4071a1c --- /dev/null +++ b/services/core/internal/protosocket/events.go @@ -0,0 +1,75 @@ +package protosocket + +import ( + "context" + + "github.com/nomadcode/nomadcode-core/internal/notification" +) + +// SuccessResponse builds a response envelope correlated to a request. +func SuccessResponse(req Envelope, payload map[string]any) Envelope { + return Envelope{ + ProtocolVersion: ProtocolVersion, + ID: generateID(), + CorrelationID: req.ID, + Type: "response", + Channel: req.Channel, + Action: req.Action, + Payload: payload, + } +} + +// ErrorResponse builds an error envelope correlated to a request. +func ErrorResponse(req Envelope, code, message string, retryable bool) Envelope { + return Envelope{ + ProtocolVersion: ProtocolVersion, + ID: generateID(), + CorrelationID: req.ID, + Type: "error", + Channel: req.Channel, + Action: req.Action, + Error: &EnvelopeError{ + Code: code, + Message: message, + Retryable: retryable, + }, + } +} + +// Broadcaster publishes an envelope to all connected clients. +type Broadcaster interface { + BroadcastEnvelope(ctx context.Context, env Envelope) error +} + +// TaskEventBroadcaster adapts notification task events into proto-socket +// task.status.changed events. It implements notification.TaskEventSink. +type TaskEventBroadcaster struct { + broadcaster Broadcaster +} + +func NewTaskEventBroadcaster(broadcaster Broadcaster) *TaskEventBroadcaster { + return &TaskEventBroadcaster{broadcaster: broadcaster} +} + +func (b *TaskEventBroadcaster) HandleTaskEvent(ctx context.Context, event notification.TaskEvent) error { + return b.broadcaster.BroadcastEnvelope(ctx, TaskStatusChangedEnvelope(event)) +} + +// TaskStatusChangedEnvelope maps a notification task event onto the +// task.status.changed event envelope. +func TaskStatusChangedEnvelope(event notification.TaskEvent) Envelope { + return Envelope{ + ProtocolVersion: ProtocolVersion, + ID: generateID(), + Type: "event", + Channel: "task", + Action: "task.status.changed", + Payload: map[string]any{ + "id": event.TaskID, + "type": string(event.Type), + "status": event.Status, + "title": event.Title, + "message": event.Message, + }, + } +} diff --git a/services/core/internal/protosocket/server.go b/services/core/internal/protosocket/server.go new file mode 100644 index 0000000..0328083 --- /dev/null +++ b/services/core/internal/protosocket/server.go @@ -0,0 +1,153 @@ +package protosocket + +import ( + "context" + "log/slog" + "net/http" + "sync" + + toki "git.toki-labs.com/toki/proto-socket/go" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/structpb" + "nhooyr.io/websocket" +) + +type Config struct { + HeartbeatIntervalSec int + HeartbeatWaitSec int +} + +type Server struct { + cfg Config + dispatcher *Dispatcher + logger *slog.Logger + mu sync.Mutex + clients map[*toki.WsClient]struct{} +} + +func NewServer(cfg Config, logger *slog.Logger) *Server { + return &Server{ + cfg: cfg, + dispatcher: NewDispatcher(), + logger: logger, + clients: make(map[*toki.WsClient]struct{}), + } +} + +func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { + opts := &websocket.AcceptOptions{ + InsecureSkipVerify: true, + } + conn, err := websocket.Accept(w, r, opts) + if err != nil { + s.logger.Error("failed to accept websocket connection", "error", err) + return + } + + client := toki.NewWsClient( + conn, + s.cfg.HeartbeatIntervalSec, + s.cfg.HeartbeatWaitSec, + ParserMap(), + ) + + s.mu.Lock() + s.clients[client] = struct{}{} + s.mu.Unlock() + + client.AddDisconnectListener(func(c *toki.WsClient) { + s.mu.Lock() + delete(s.clients, c) + s.mu.Unlock() + s.logger.Info("websocket client disconnected") + }) + + toki.AddRequestListenerTyped[*structpb.Struct, *structpb.Struct](&client.Communicator, func(req *structpb.Struct) (*structpb.Struct, error) { + env, err := EnvelopeFromStruct(req) + if err != nil { + errEnv := Envelope{ + ProtocolVersion: ProtocolVersion, + ID: generateID(), + CorrelationID: "", + Type: "error", + Error: &EnvelopeError{ + Code: "INVALID_ENVELOPE", + Message: "failed to parse envelope: " + err.Error(), + }, + } + res, _ := errEnv.ToStruct() + return res, nil + } + + resEnv := s.dispatcher.Dispatch(r.Context(), env) + resStruct, err := resEnv.ToStruct() + if err != nil { + errEnv := Envelope{ + ProtocolVersion: ProtocolVersion, + ID: generateID(), + CorrelationID: env.ID, + Type: "error", + Error: &EnvelopeError{ + Code: "SERIALIZATION_ERROR", + Message: "failed to serialize response envelope: " + err.Error(), + }, + } + res, _ := errEnv.ToStruct() + return res, nil + } + + return resStruct, nil + }) + + s.logger.Info("websocket client connected successfully") +} + +func (s *Server) Close() error { + s.mu.Lock() + clients := make([]*toki.WsClient, 0, len(s.clients)) + for cl := range s.clients { + clients = append(clients, cl) + } + s.mu.Unlock() + + for _, cl := range clients { + _ = cl.Close() + } + return nil +} + +func (s *Server) BroadcastEnvelope(ctx context.Context, env Envelope) error { + structMsg, err := env.ToStruct() + if err != nil { + return err + } + + s.mu.Lock() + clients := make([]*toki.WsClient, 0, len(s.clients)) + for cl := range s.clients { + clients = append(clients, cl) + } + s.mu.Unlock() + + for _, cl := range clients { + if cl.IsAlive() { + if err := cl.Send(structMsg); err != nil { + s.logger.Error("failed to send broadcast to client", "error", err) + } + } + } + return nil +} + +func (s *Server) Dispatcher() *Dispatcher { + return s.dispatcher +} + +func ParserMap() toki.ParserMap { + return toki.ParserMap{ + toki.TypeNameOf(&structpb.Struct{}): func(b []byte) (proto.Message, error) { + msg := &structpb.Struct{} + return msg, proto.Unmarshal(b, msg) + }, + } +} diff --git a/services/core/internal/protosocket/server_test.go b/services/core/internal/protosocket/server_test.go new file mode 100644 index 0000000..f565957 --- /dev/null +++ b/services/core/internal/protosocket/server_test.go @@ -0,0 +1,138 @@ +package protosocket + +import ( + "context" + "io" + "log/slog" + "net/http/httptest" + "strings" + "testing" + "time" + + toki "git.toki-labs.com/toki/proto-socket/go" + "google.golang.org/protobuf/types/known/structpb" + "nhooyr.io/websocket" +) + +func TestServerAcceptsWebSocketAndClosesClients(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + srv := NewServer(Config{ + HeartbeatIntervalSec: 0, + HeartbeatWaitSec: 0, + }, logger) + + testServer := httptest.NewServer(srv) + defer testServer.Close() + + wsURL := strings.Replace(testServer.URL, "http", "ws", 1) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + conn, _, err := websocket.Dial(ctx, wsURL, nil) + if err != nil { + t.Fatalf("failed to dial websocket: %v", err) + } + defer conn.Close(websocket.StatusNormalClosure, "") + + client := toki.NewWsClient(conn, 0, 0, ParserMap()) + defer client.Close() + + if !client.IsAlive() { + t.Error("client should be alive after connection") + } + + disconnected := make(chan struct{}, 1) + client.AddDisconnectListener(func(c *toki.WsClient) { + disconnected <- struct{}{} + }) + + if err := srv.Close(); err != nil { + t.Errorf("failed to close server: %v", err) + } + + select { + case <-disconnected: + // success + case <-time.After(1 * time.Second): + t.Error("timed out waiting for client to disconnect after server close") + } +} + +func TestServerRespondsWithErrorEnvelopeForUnsupportedAction(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + srv := NewServer(Config{ + HeartbeatIntervalSec: 0, + HeartbeatWaitSec: 0, + }, logger) + defer srv.Close() + + testServer := httptest.NewServer(srv) + defer testServer.Close() + + wsURL := strings.Replace(testServer.URL, "http", "ws", 1) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + conn, _, err := websocket.Dial(ctx, wsURL, nil) + if err != nil { + t.Fatalf("failed to dial websocket: %v", err) + } + defer conn.Close(websocket.StatusNormalClosure, "") + + client := toki.NewWsClient(conn, 0, 0, ParserMap()) + defer client.Close() + + requestEnv := Envelope{ + ProtocolVersion: ProtocolVersion, + ID: "req-999", + Type: "request", + Channel: "tasks", + Action: "invalid-unsupported-action", + } + reqStruct, err := requestEnv.ToStruct() + if err != nil { + t.Fatalf("failed to create request struct: %v", err) + } + + resStruct, err := toki.SendRequestTyped[*structpb.Struct, *structpb.Struct](&client.Communicator, reqStruct, 1*time.Second) + if err != nil { + t.Fatalf("failed to send request: %v", err) + } + + responseEnv, err := EnvelopeFromStruct(resStruct) + if err != nil { + t.Fatalf("failed to parse response struct: %v", err) + } + + if responseEnv.Type != "error" { + t.Errorf("expected response type 'error', got %q", responseEnv.Type) + } + if responseEnv.CorrelationID != "req-999" { + t.Errorf("expected CorrelationID 'req-999', got %q", responseEnv.CorrelationID) + } + if responseEnv.Error == nil { + t.Fatal("expected error in response env") + } + if responseEnv.Error.Code != "UNSUPPORTED_ACTION" { + t.Errorf("expected error code 'UNSUPPORTED_ACTION', got %q", responseEnv.Error.Code) + } + if responseEnv.Error.Retryable { + t.Error("expected Retryable to be false for UNSUPPORTED_ACTION") + } +} + +func TestServerHeartbeatConfigSupport(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + + srvZero := NewServer(Config{ + HeartbeatIntervalSec: 0, + HeartbeatWaitSec: 0, + }, logger) + srvZero.Close() + + srvNonZero := NewServer(Config{ + HeartbeatIntervalSec: 10, + HeartbeatWaitSec: 2, + }, logger) + srvNonZero.Close() +} diff --git a/services/core/internal/protosocket/tasks.go b/services/core/internal/protosocket/tasks.go new file mode 100644 index 0000000..725a715 --- /dev/null +++ b/services/core/internal/protosocket/tasks.go @@ -0,0 +1,182 @@ +package protosocket + +import ( + "context" + "encoding/json" + "errors" + + "github.com/jackc/pgx/v5" + + "github.com/nomadcode/nomadcode-core/internal/storage" + "github.com/nomadcode/nomadcode-core/internal/workflow" +) + +// TaskService is the subset of the workflow service required to map REST task +// semantics onto proto-socket actions. +type TaskService interface { + CreateTask(context.Context, workflow.CreateTaskInput) (storage.Task, error) + ListTasks(context.Context, int32) ([]storage.Task, error) + GetTask(context.Context, string) (storage.Task, error) + EnqueueTask(context.Context, string) (storage.Task, error) +} + +// TaskChannels registers task.* request handlers on a dispatcher. +type TaskChannels struct { + tasks TaskService +} + +func NewTaskChannels(tasks TaskService) *TaskChannels { + return &TaskChannels{tasks: tasks} +} + +func (c *TaskChannels) Register(dispatcher *Dispatcher) { + dispatcher.Register("task.create", c.handleCreate) + dispatcher.Register("task.list", c.handleList) + dispatcher.Register("task.get", c.handleGet) + dispatcher.Register("task.enqueue", c.handleEnqueue) +} + +type listRequest struct { + Limit int32 `json:"limit"` +} + +type idRequest struct { + ID string `json:"id"` +} + +func (c *TaskChannels) handleCreate(ctx context.Context, req Envelope) Envelope { + input, err := payloadAs[workflow.CreateTaskInput](req.Payload) + if err != nil { + return invalidPayloadResponse(req, err) + } + + task, err := c.tasks.CreateTask(ctx, input) + if err != nil { + return mapTaskError(req, err) + } + + taskMap, err := taskToMap(task) + if err != nil { + return ErrorResponse(req, "internal.error", "internal server error", true) + } + + return SuccessResponse(req, map[string]any{ + "id": task.ID, + "status": task.Status, + "external_provider": stringOrEmpty(task.ExternalProvider), + "external_id": stringOrEmpty(task.ExternalID), + "task": taskMap, + }) +} + +func (c *TaskChannels) handleList(ctx context.Context, req Envelope) Envelope { + input, err := payloadAs[listRequest](req.Payload) + if err != nil { + return invalidPayloadResponse(req, err) + } + + tasks, err := c.tasks.ListTasks(ctx, input.Limit) + if err != nil { + return mapTaskError(req, err) + } + + encoded := make([]any, 0, len(tasks)) + for _, task := range tasks { + taskMap, err := taskToMap(task) + if err != nil { + return ErrorResponse(req, "internal.error", "internal server error", true) + } + encoded = append(encoded, taskMap) + } + + return SuccessResponse(req, map[string]any{"tasks": encoded}) +} + +func (c *TaskChannels) handleGet(ctx context.Context, req Envelope) Envelope { + input, err := payloadAs[idRequest](req.Payload) + if err != nil { + return invalidPayloadResponse(req, err) + } + + task, err := c.tasks.GetTask(ctx, input.ID) + if err != nil { + return mapTaskError(req, err) + } + + taskMap, err := taskToMap(task) + if err != nil { + return ErrorResponse(req, "internal.error", "internal server error", true) + } + + return SuccessResponse(req, map[string]any{"task": taskMap}) +} + +func (c *TaskChannels) handleEnqueue(ctx context.Context, req Envelope) Envelope { + input, err := payloadAs[idRequest](req.Payload) + if err != nil { + return invalidPayloadResponse(req, err) + } + + task, err := c.tasks.EnqueueTask(ctx, input.ID) + if err != nil { + return mapTaskError(req, err) + } + + taskMap, err := taskToMap(task) + if err != nil { + return ErrorResponse(req, "internal.error", "internal server error", true) + } + + return SuccessResponse(req, map[string]any{ + "id": task.ID, + "status": task.Status, + "task": taskMap, + }) +} + +func mapTaskError(req Envelope, err error) Envelope { + switch { + case errors.Is(err, workflow.ErrInvalidTaskInput): + return ErrorResponse(req, "task.invalid_input", err.Error(), false) + case errors.Is(err, workflow.ErrTaskCannotBeEnqueued): + return ErrorResponse(req, "task.conflict", err.Error(), false) + case errors.Is(err, pgx.ErrNoRows): + return ErrorResponse(req, "task.not_found", "task not found", false) + default: + return ErrorResponse(req, "internal.error", "internal server error", true) + } +} + +func invalidPayloadResponse(req Envelope, err error) Envelope { + return ErrorResponse(req, "task.invalid_payload", "invalid payload: "+err.Error(), false) +} + +// payloadAs decodes an envelope payload into a typed request by round-tripping +// through JSON instead of ad hoc map indexing. +func payloadAs[T any](payload map[string]any) (T, error) { + var out T + b, err := json.Marshal(payload) + if err != nil { + return out, err + } + return out, json.Unmarshal(b, &out) +} + +func taskToMap(task storage.Task) (map[string]any, error) { + b, err := json.Marshal(task) + if err != nil { + return nil, err + } + var out map[string]any + if err := json.Unmarshal(b, &out); err != nil { + return nil, err + } + return out, nil +} + +func stringOrEmpty(value *string) string { + if value == nil { + return "" + } + return *value +} diff --git a/services/core/internal/protosocket/tasks_test.go b/services/core/internal/protosocket/tasks_test.go new file mode 100644 index 0000000..0fcab6c --- /dev/null +++ b/services/core/internal/protosocket/tasks_test.go @@ -0,0 +1,448 @@ +package protosocket + +import ( + "context" + "encoding/base64" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + toki "git.toki-labs.com/toki/proto-socket/go" + "github.com/jackc/pgx/v5" + "google.golang.org/protobuf/types/known/structpb" + "nhooyr.io/websocket" + + apphttp "github.com/nomadcode/nomadcode-core/internal/http" + "github.com/nomadcode/nomadcode-core/internal/notification" + "github.com/nomadcode/nomadcode-core/internal/storage" + "github.com/nomadcode/nomadcode-core/internal/workflow" +) + +type fakeTaskService struct { + created workflow.CreateTaskInput + createOut storage.Task + createErr error + + listLimit int32 + listOut []storage.Task + listErr error + + getID string + getOut storage.Task + getErr error + + enqueueID string + enqueueOut storage.Task + enqueueErr error +} + +func (f *fakeTaskService) CreateTask(_ context.Context, in workflow.CreateTaskInput) (storage.Task, error) { + f.created = in + return f.createOut, f.createErr +} + +func (f *fakeTaskService) ListTasks(_ context.Context, limit int32) ([]storage.Task, error) { + f.listLimit = limit + return f.listOut, f.listErr +} + +func (f *fakeTaskService) GetTask(_ context.Context, id string) (storage.Task, error) { + f.getID = id + return f.getOut, f.getErr +} + +func (f *fakeTaskService) EnqueueTask(_ context.Context, id string) (storage.Task, error) { + f.enqueueID = id + return f.enqueueOut, f.enqueueErr +} + +func dispatchTo(fake TaskService, req Envelope) Envelope { + d := NewDispatcher() + NewTaskChannels(fake).Register(d) + return d.Dispatch(context.Background(), req) +} + +func TestTaskCreateActionCallsWorkflowAndReturnsResponse(t *testing.T) { + fake := &fakeTaskService{ + createOut: storage.Task{ID: "task-1", Status: "pending", Title: "My Task", Source: "client"}, + } + + res := dispatchTo(fake, Envelope{ + ID: "r1", + Type: "request", + Channel: "task", + Action: "task.create", + Payload: map[string]any{ + "title": "My Task", + "source": "client", + "payload": map[string]any{}, + }, + }) + + if fake.created.Title != "My Task" || fake.created.Source != "client" { + t.Fatalf("workflow not called with expected input: %+v", fake.created) + } + if res.Type != "response" { + t.Fatalf("expected response type, got %q (err=%+v)", res.Type, res.Error) + } + if res.CorrelationID != "r1" { + t.Errorf("expected correlation_id r1, got %q", res.CorrelationID) + } + if res.Payload["id"] != "task-1" || res.Payload["status"] != "pending" { + t.Errorf("unexpected response payload: %+v", res.Payload) + } + if _, ok := res.Payload["task"].(map[string]any); !ok { + t.Errorf("expected nested task map in response payload: %+v", res.Payload) + } +} + +func TestTaskListActionForwardsLimitToService(t *testing.T) { + fake := &fakeTaskService{ + listOut: []storage.Task{{ID: "task-1"}, {ID: "task-2"}}, + } + + res := dispatchTo(fake, Envelope{ + ID: "r2", + Type: "request", + Channel: "task", + Action: "task.list", + Payload: map[string]any{"limit": float64(5)}, + }) + + if fake.listLimit != 5 { + t.Errorf("expected limit 5 forwarded to service, got %d", fake.listLimit) + } + if res.Type != "response" { + t.Fatalf("expected response type, got %q", res.Type) + } + tasks, ok := res.Payload["tasks"].([]any) + if !ok || len(tasks) != 2 { + t.Errorf("expected 2 tasks in payload, got %+v", res.Payload["tasks"]) + } + + // Omitted limit forwards 0; the workflow service owns default/bounding. + resDefault := dispatchTo(fake, Envelope{ID: "r2b", Action: "task.list", Channel: "task", Payload: map[string]any{}}) + if fake.listLimit != 0 { + t.Errorf("expected omitted limit to forward 0, got %d", fake.listLimit) + } + if resDefault.Type != "response" { + t.Errorf("expected response type for default limit, got %q", resDefault.Type) + } +} + +func TestTaskGetActionMapsNotFoundError(t *testing.T) { + fake := &fakeTaskService{getErr: pgx.ErrNoRows} + + res := dispatchTo(fake, Envelope{ + ID: "r3", + Action: "task.get", + Channel: "task", + Payload: map[string]any{"id": "missing"}, + }) + + if fake.getID != "missing" { + t.Errorf("expected get called with 'missing', got %q", fake.getID) + } + if res.Type != "error" { + t.Fatalf("expected error type, got %q", res.Type) + } + if res.Error == nil || res.Error.Code != "task.not_found" { + t.Errorf("expected task.not_found, got %+v", res.Error) + } + if res.Error.Retryable { + t.Error("expected not_found to be non-retryable") + } +} + +func TestTaskEnqueueActionMapsConflictError(t *testing.T) { + fake := &fakeTaskService{enqueueErr: workflow.ErrTaskCannotBeEnqueued} + + res := dispatchTo(fake, Envelope{ + ID: "r4", + Action: "task.enqueue", + Channel: "task", + Payload: map[string]any{"id": "task-9"}, + }) + + if res.Type != "error" { + t.Fatalf("expected error type, got %q", res.Type) + } + if res.Error == nil || res.Error.Code != "task.conflict" { + t.Errorf("expected task.conflict, got %+v", res.Error) + } +} + +func TestTaskActionRejectsInvalidPayload(t *testing.T) { + fake := &fakeTaskService{} + + res := dispatchTo(fake, Envelope{ + ID: "r5", + Action: "task.get", + Channel: "task", + Payload: map[string]any{"id": float64(123)}, // wrong type for string id + }) + + if res.Type != "error" { + t.Fatalf("expected error type, got %q", res.Type) + } + if res.Error == nil || res.Error.Code != "task.invalid_payload" { + t.Errorf("expected task.invalid_payload, got %+v", res.Error) + } + if fake.getID != "" { + t.Errorf("service should not be called on invalid payload, got %q", fake.getID) + } +} + +type captureBroadcaster struct { + env Envelope + err error +} + +func (c *captureBroadcaster) BroadcastEnvelope(_ context.Context, env Envelope) error { + c.env = env + return c.err +} + +func TestHandleTaskEventBroadcastsStatusChangedEnvelope(t *testing.T) { + cap := &captureBroadcaster{} + sink := NewTaskEventBroadcaster(cap) + + err := sink.HandleTaskEvent(context.Background(), notification.TaskEvent{ + Type: notification.TaskEventRunning, + TaskID: "task-7", + Title: "Run me", + Status: "running", + Message: "started", + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if cap.env.Type != "event" || cap.env.Channel != "task" || cap.env.Action != "task.status.changed" { + t.Fatalf("unexpected envelope: type=%q channel=%q action=%q", cap.env.Type, cap.env.Channel, cap.env.Action) + } + if cap.env.Payload["id"] != "task-7" || cap.env.Payload["status"] != "running" { + t.Errorf("unexpected event payload: %+v", cap.env.Payload) + } + if cap.env.Payload["type"] != string(notification.TaskEventRunning) { + t.Errorf("expected type %q, got %v", notification.TaskEventRunning, cap.env.Payload["type"]) + } +} + +// --- route-level contract tests over the real router + proto-socket server --- + +func newContractRouter(t *testing.T, fake TaskService) (*httptest.Server, func()) { + t.Helper() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + srv := NewServer(Config{HeartbeatIntervalSec: 0, HeartbeatWaitSec: 0}, logger) + NewTaskChannels(fake).Register(srv.Dispatcher()) + + router := apphttp.NewRouter( + apphttp.NewHandler(nil, nil, nil, logger), + logger, + apphttp.AuthConfig{Username: "nomadcode", Password: "secret"}, + srv, + "/proto-socket", + ) + ts := httptest.NewServer(router) + return ts, func() { + ts.Close() + _ = srv.Close() + } +} + +func wsURL(ts *httptest.Server) string { + return strings.Replace(ts.URL, "http", "ws", 1) + "/proto-socket" +} + +func authDialOptions() *websocket.DialOptions { + header := http.Header{} + header.Set("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte("nomadcode:secret"))) + return &websocket.DialOptions{HTTPHeader: header} +} + +func TestRouterProtoSocketTaskListRequiresAuth(t *testing.T) { + fake := &fakeTaskService{listOut: []storage.Task{{ID: "task-1"}}} + ts, cleanup := newContractRouter(t, fake) + defer cleanup() + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + // Unauthenticated dial must fail at the route. + if conn, _, err := websocket.Dial(ctx, wsURL(ts), nil); err == nil { + conn.Close(websocket.StatusNormalClosure, "") + t.Fatal("expected unauthenticated websocket dial to fail") + } + + // Authenticated dial succeeds and task.list returns a response envelope. + conn, _, err := websocket.Dial(ctx, wsURL(ts), authDialOptions()) + if err != nil { + t.Fatalf("authenticated dial failed: %v", err) + } + defer conn.Close(websocket.StatusNormalClosure, "") + + client := toki.NewWsClient(conn, 0, 0, ParserMap()) + defer client.Close() + + reqStruct, err := Envelope{ + ProtocolVersion: ProtocolVersion, + ID: "req-list", + Type: "request", + Channel: "task", + Action: "task.list", + Payload: map[string]any{"limit": float64(20)}, + }.ToStruct() + if err != nil { + t.Fatalf("failed to build request: %v", err) + } + + resStruct, err := toki.SendRequestTyped[*structpb.Struct, *structpb.Struct](&client.Communicator, reqStruct, 2*time.Second) + if err != nil { + t.Fatalf("send request failed: %v", err) + } + res, err := EnvelopeFromStruct(resStruct) + if err != nil { + t.Fatalf("parse response failed: %v", err) + } + if res.Type != "response" || res.CorrelationID != "req-list" { + t.Fatalf("unexpected response: type=%q correlation=%q", res.Type, res.CorrelationID) + } + if _, ok := res.Payload["tasks"]; !ok { + t.Errorf("expected tasks payload, got %+v", res.Payload) + } +} + +func TestProtoSocketTaskActionsMatchRestSemantics(t *testing.T) { + fake := &fakeTaskService{ + createOut: storage.Task{ID: "task-c", Status: "pending"}, + getOut: storage.Task{ID: "task-g", Status: "running"}, + enqueueOut: storage.Task{ID: "task-e", Status: "queued"}, + } + ts, cleanup := newContractRouter(t, fake) + defer cleanup() + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + conn, _, err := websocket.Dial(ctx, wsURL(ts), authDialOptions()) + if err != nil { + t.Fatalf("dial failed: %v", err) + } + defer conn.Close(websocket.StatusNormalClosure, "") + client := toki.NewWsClient(conn, 0, 0, ParserMap()) + defer client.Close() + + send := func(action string, payload map[string]any) Envelope { + reqStruct, err := Envelope{ + ProtocolVersion: ProtocolVersion, + ID: "req-" + action, + Type: "request", + Channel: "task", + Action: action, + Payload: payload, + }.ToStruct() + if err != nil { + t.Fatalf("build request %s: %v", action, err) + } + resStruct, err := toki.SendRequestTyped[*structpb.Struct, *structpb.Struct](&client.Communicator, reqStruct, 2*time.Second) + if err != nil { + t.Fatalf("send %s: %v", action, err) + } + res, err := EnvelopeFromStruct(resStruct) + if err != nil { + t.Fatalf("parse %s: %v", action, err) + } + return res + } + + create := send("task.create", map[string]any{"title": "T", "source": "client", "payload": map[string]any{}}) + if create.Type != "response" || create.Payload["id"] != "task-c" || create.Payload["status"] != "pending" { + t.Errorf("unexpected create response: %+v (err=%+v)", create.Payload, create.Error) + } + + get := send("task.get", map[string]any{"id": "task-g"}) + if get.Type != "response" { + t.Errorf("unexpected get response type: %q", get.Type) + } + if task, ok := get.Payload["task"].(map[string]any); !ok || task["status"] != "running" { + t.Errorf("unexpected get task payload: %+v", get.Payload) + } + + enqueue := send("task.enqueue", map[string]any{"id": "task-e"}) + if enqueue.Type != "response" || enqueue.Payload["status"] != "queued" { + t.Errorf("unexpected enqueue response: %+v", enqueue.Payload) + } +} + +func TestProtoSocketTaskErrorEnvelopeCodes(t *testing.T) { + fake := &fakeTaskService{ + getErr: pgx.ErrNoRows, + enqueueErr: workflow.ErrTaskCannotBeEnqueued, + createErr: workflow.ErrInvalidTaskInput, + } + ts, cleanup := newContractRouter(t, fake) + defer cleanup() + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + conn, _, err := websocket.Dial(ctx, wsURL(ts), authDialOptions()) + if err != nil { + t.Fatalf("dial failed: %v", err) + } + defer conn.Close(websocket.StatusNormalClosure, "") + client := toki.NewWsClient(conn, 0, 0, ParserMap()) + defer client.Close() + + send := func(action string, payload map[string]any) Envelope { + reqStruct, _ := Envelope{ProtocolVersion: ProtocolVersion, ID: "e-" + action, Type: "request", Channel: "task", Action: action, Payload: payload}.ToStruct() + resStruct, err := toki.SendRequestTyped[*structpb.Struct, *structpb.Struct](&client.Communicator, reqStruct, 2*time.Second) + if err != nil { + t.Fatalf("send %s: %v", action, err) + } + res, _ := EnvelopeFromStruct(resStruct) + return res + } + + cases := []struct { + action string + payload map[string]any + code string + }{ + {"task.get", map[string]any{"id": "x"}, "task.not_found"}, + {"task.enqueue", map[string]any{"id": "x"}, "task.conflict"}, + {"task.create", map[string]any{"title": "T", "source": "client"}, "task.invalid_input"}, + {"task.get", map[string]any{"id": float64(1)}, "task.invalid_payload"}, + {"task.unknown", map[string]any{}, "UNSUPPORTED_ACTION"}, + } + for _, tc := range cases { + res := send(tc.action, tc.payload) + if res.Type != "error" { + t.Errorf("%s: expected error type, got %q", tc.action, res.Type) + continue + } + if res.Error == nil || res.Error.Code != tc.code { + t.Errorf("%s: expected code %q, got %+v", tc.action, tc.code, res.Error) + } + } +} + +func TestProtoSocketServerUsesConfiguredHeartbeat(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + srv := NewServer(Config{HeartbeatIntervalSec: 1, HeartbeatWaitSec: 1}, logger) + defer srv.Close() + + ts := httptest.NewServer(srv) + defer ts.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + conn, _, err := websocket.Dial(ctx, strings.Replace(ts.URL, "http", "ws", 1), nil) + if err != nil { + t.Fatalf("dial failed: %v", err) + } + conn.Close(websocket.StatusNormalClosure, "") +}