From 68ff677f863d2411c21d49e57c42617f2844ed4f Mon Sep 17 00:00:00 2001 From: Vitalii Valkov Date: Sat, 1 Aug 2026 02:18:30 +0200 Subject: [PATCH] feat: resolve an omitted thread from the caller's instance An agent CLI's final turn text had no way to say where it belonged: the 2026-07-14 change removed THREAD_ID and replaced it with per-send explicit targeting, which a wrapped CLI cannot express. An agent instance may now leave thread_id out and have it resolved from its own default_thread_id. Resolution is server-side from the caller identity, so nothing the container carries can misroute a message -- that was the objection to the environment variable, and it still holds. Every other caller type, and an instance with no default, is refused. Deliberately not the thread of whatever woke the instance: in A -> B -> C, B is woken by C's reply on thread BC while owing its answer to A on thread AB. Origin composes because delegation creates sub-threads downward; the trigger does not. Class-on-add now reports the thread doing the adding as a creation circumstance for the class policy to act on. CreateThread settles the thread id before finalizing participants, because creating an instance has to name the thread that is adding it. --- .devspace/cache.yaml | 3 + .devspace/logs/default.log | 19 ++++ .devspace/logs/dev.dev:threads.log | 92 ++++++++++++++++++++ .devspace/logs/dev.threads.log | 7 ++ .devspace/logs/errors.log | 2 + .devspace/logs/portforwarding.log | 2 + internal/server/default_thread_test.go | 115 +++++++++++++++++++++++++ internal/server/server.go | 88 ++++++++++++++++--- internal/server/server_test.go | 30 +++---- internal/store/threads.go | 7 +- 10 files changed, 336 insertions(+), 29 deletions(-) create mode 100644 .devspace/cache.yaml create mode 100644 .devspace/logs/default.log create mode 100644 .devspace/logs/dev.dev:threads.log create mode 100644 .devspace/logs/dev.threads.log create mode 100644 .devspace/logs/errors.log create mode 100644 .devspace/logs/portforwarding.log create mode 100644 internal/server/default_thread_test.go diff --git a/.devspace/cache.yaml b/.devspace/cache.yaml new file mode 100644 index 0000000..7546f14 --- /dev/null +++ b/.devspace/cache.yaml @@ -0,0 +1,3 @@ +lastContext: + namespace: default + context: agyn-local diff --git a/.devspace/logs/default.log b/.devspace/logs/default.log new file mode 100644 index 0000000..8e8528c --- /dev/null +++ b/.devspace/logs/default.log @@ -0,0 +1,19 @@ +{"level":"info","msg":"Using namespace 'default'","time":"2026-08-01T00:01:40+02:00"} +{"level":"info","msg":"Using kube context 'agyn-local'","time":"2026-08-01T00:01:40+02:00"} +{"level":"error","msg":"pipelines.test:e2e has to match the following regex: ^(([a-z0-9][a-z0-9\\-]*[a-z0-9])|([a-z0-9]))$","time":"2026-08-01T00:01:40+02:00"} +{"level":"info","msg":"Using namespace 'default'","time":"2026-08-01T00:13:08+02:00"} +{"level":"info","msg":"Using kube context 'agyn-local'","time":"2026-08-01T00:13:08+02:00"} +WARNING: ArgoCD Application 'threads' not found in argocd namespace.Patching threads deployment for DevSpace...deployment.apps/threads patcheddev:threads Waiting for pod to become ready...dev:threads Pod threads-64c9c4cb6d-6zp7k: 0/1 nodes are available: 1 Insufficient cpu, 1 Insufficient memory. preemption: 0/1 nodes are available: 1 No preemption victims found for incoming pod. (FailedScheduling)dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads Pod threads-64c9c4cb6d-6zp7k: 0/1 nodes are available: 1 Insufficient cpu, 1 Insufficient memory. preemption: 0/1 nodes are available: 1 No preemption victims found for incoming pod. (FailedScheduling)dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutdev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeoutstart_dev: waiting for pod to become ready: timeout: couldn't find a pod / container in time with label selector: app.kubernetes.io/instance=threads,app.kubernetes.io/name=threads{"level":"error","msg":"exit status 1","time":"2026-08-01T00:23:08+02:00"} +{"level":"info","msg":"Using namespace 'default'","time":"2026-08-01T01:05:15+02:00"} +{"level":"info","msg":"Using kube context 'agyn-local'","time":"2026-08-01T01:05:15+02:00"} +WARNING: ArgoCD Application 'threads' not found in argocd namespace.Patching threads deployment for DevSpace...deployment.apps/threads patcheddev:threads Waiting for pod to become ready...dev:threads Pod threads-64c9c4cb6d-jfzzz: Liveness probe failed: dial tcp 10.42.0.6:50051: connect: connection refused (Unhealthy)dev:threads Pod threads-64c9c4cb6d-jfzzz: Readiness probe failed: dial tcp 10.42.0.6:50051: connect: connection refused (Unhealthy){"level":"info","msg":"Using namespace 'default'","time":"2026-08-01T01:10:27+02:00"} +{"level":"info","msg":"Using kube context 'agyn-local'","time":"2026-08-01T01:10:27+02:00"} +WARNING: ArgoCD Application 'threads' not found in argocd namespace.Patching threads deployment for DevSpace...deployment.apps/threads patchedstart_dev: couldn't execute 'devspace dev', because there is another DevSpace session for the project (threads) already running inside this namespaceYou may want to use one of these commands instead:- devspace enter: opens a terminal session for a container- devspace attach: attaches to the PID 1 process (entrypoint) of a container- devspace logs: streams the logs of a container- devspace sync: syncs files between your local filesyste and a container's filesystem- devspace ui: starts the DevSpace localhost UI{"level":"error","msg":"exit status 1","time":"2026-08-01T01:10:27+02:00"} +dev:threads Pod threads-5dcf65b76d-dwhk8: Readiness probe failed: dial tcp 10.42.0.14:50051: connect: connection refused (Unhealthy)dev:threads Pod threads-5dcf65b76d-dwhk8 has critical status: Error. DevSpace will continue waiting, but this operation might timeoutdev:threads Pod threads-5dcf65b76d-dwhk8 has critical status: CrashLoopBackOff. DevSpace will continue waiting, but this operation might timeoutdev:threads Pod threads-5dcf65b76d-dwhk8: Back-off restarting failed container threads in pod threads-5dcf65b76d-dwhk8_platform(86b305a0-80cb-40df-b6a1-2223d003d7bb) (BackOff)start_dev: waiting for pod to become ready: timeout: couldn't find a pod / container in time with label selector: app.kubernetes.io/name=threads{"level":"error","msg":"exit status 1","time":"2026-08-01T01:15:16+02:00"} +{"level":"info","msg":"Using namespace 'default'","time":"2026-08-01T01:17:50+02:00"} +{"level":"info","msg":"Using kube context 'agyn-local'","time":"2026-08-01T01:17:50+02:00"} +WARNING: ArgoCD Application 'threads' not found in argocd namespace.Patching threads deployment for DevSpace...deployment.apps/threads patched (no change)dev:threads Waiting for pod to become ready...dev:threads Pod threads-5dcf65b76d-dwhk8: Back-off restarting failed container threads in pod threads-5dcf65b76d-dwhk8_platform(86b305a0-80cb-40df-b6a1-2223d003d7bb) (BackOff)dev:threads Pod threads-5dcf65b76d-dwhk8: Readiness probe failed: dial tcp 10.42.0.14:50051: connect: connection refused (Unhealthy){"level":"info","msg":"Using namespace 'default'","time":"2026-08-01T01:19:08+02:00"} +{"level":"info","msg":"Using kube context 'agyn-local'","time":"2026-08-01T01:19:08+02:00"} +WARNING: ArgoCD Application 'threads' not found in argocd namespace.Patching threads deployment for DevSpace...deployment.apps/threads patcheddev:threads Waiting for pod to become ready...dev:threads Pod threads-5b7cf54c7c-q4d7c: Readiness probe failed: dial tcp 10.42.0.15:50051: connect: connection refused (Unhealthy){"level":"info","msg":"Using namespace 'default'","time":"2026-08-01T01:25:24+02:00"} +{"level":"info","msg":"Using kube context 'agyn-local'","time":"2026-08-01T01:25:24+02:00"} +WARNING: ArgoCD Application 'threads' not found in argocd namespace.Patching threads deployment for DevSpace...deployment.apps/threads patcheddev:threads Waiting for pod to become ready...dev:threads Selected pod threads-68dd7fb865-8jsb5dev:threads ports Port forwarding started on: 50051 -> 50051dev:threads sync Sync started on: ./ <-> /opt/app/datadev:threads sync Waiting for initial sync to completedev:threads sync Initial sync completedWaiting for threads to be healthy on :50051...exec_container: command terminated with exit code 1dev:threads logs Streaming logs of pod:container threads-68dd7fb865-8jsb5:threadsdev:threads logs threads: go: downloading github.com/jackc/pgx/v5 v5.8.0dev:threads logs threads: go: downloading google.golang.org/grpc v1.79.1exec_container: command terminated with exit code 1dev:threads logs threads: go: downloading google.golang.org/protobuf v1.36.11dev:threads logs threads: go: downloading github.com/google/uuid v1.6.0dev:threads logs threads: go: downloading github.com/jackc/puddle/v2 v2.2.2dev:threads logs threads: go: downloading golang.org/x/net v0.48.0dev:threads logs threads: go: downloading google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217dev:threads logs threads: go: downloading github.com/jackc/pgpassfile v1.0.0dev:threads logs threads: go: downloading github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761dev:threads logs threads: go: downloading golang.org/x/text v0.32.0dev:threads logs threads: go: downloading golang.org/x/sync v0.19.0dev:threads logs threads: go: downloading golang.org/x/sys v0.39.0exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1exec_container: command terminated with exit code 1dev:threads logs threads: 2026/07/31 23:26:17 ThreadsService listening on :50051Dev environment ready. Stopping dev session.dev:threads Stopping dev threads \ No newline at end of file diff --git a/.devspace/logs/dev.dev:threads.log b/.devspace/logs/dev.dev:threads.log new file mode 100644 index 0000000..fd664cd --- /dev/null +++ b/.devspace/logs/dev.dev:threads.log @@ -0,0 +1,92 @@ +{"level":"info","msg":"dev:threads Waiting for pod to become ready...","time":"2026-08-01T00:13:08+02:00"} +{"level":"warning","msg":"dev:threads Pod threads-64c9c4cb6d-6zp7k: 0/1 nodes are available: 1 Insufficient cpu, 1 Insufficient memory. preemption: 0/1 nodes are available: 1 No preemption victims found for incoming pod. (FailedScheduling)","time":"2026-08-01T00:13:19+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:13:19+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:13:29+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:13:40+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:13:50+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:14:01+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:14:11+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:14:22+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:14:32+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:14:43+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:14:53+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:15:04+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:15:14+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:15:25+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:15:35+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:15:46+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:15:56+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:16:07+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:16:17+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:16:28+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:16:38+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:16:49+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:16:59+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:17:10+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:17:20+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:17:31+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:17:41+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:17:52+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:18:02+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:18:13+02:00"} +{"level":"warning","msg":"dev:threads Pod threads-64c9c4cb6d-6zp7k: 0/1 nodes are available: 1 Insufficient cpu, 1 Insufficient memory. preemption: 0/1 nodes are available: 1 No preemption victims found for incoming pod. (FailedScheduling)","time":"2026-08-01T00:18:23+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:18:23+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:18:34+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:18:44+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:18:55+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:19:05+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:19:16+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:19:26+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:19:37+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:19:47+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:19:58+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:20:08+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:20:19+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:20:29+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:20:40+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:20:50+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:21:01+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:21:11+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:21:22+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:21:32+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:21:43+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:21:53+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:22:04+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:22:14+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:22:25+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:22:35+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:22:46+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:22:56+02:00"} +{"level":"warning","msg":"dev:threads DevSpace still couldn't find any Pods that match the selector. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T00:23:07+02:00"} +{"level":"info","msg":"dev:threads Waiting for pod to become ready...","time":"2026-08-01T01:05:16+02:00"} +{"level":"warning","msg":"dev:threads Pod threads-64c9c4cb6d-jfzzz: Liveness probe failed: dial tcp 10.42.0.6:50051: connect: connection refused (Unhealthy)","time":"2026-08-01T01:05:26+02:00"} +{"level":"warning","msg":"dev:threads Pod threads-64c9c4cb6d-jfzzz: Readiness probe failed: dial tcp 10.42.0.6:50051: connect: connection refused (Unhealthy)","time":"2026-08-01T01:05:37+02:00"} +{"level":"warning","msg":"dev:threads Pod threads-5dcf65b76d-dwhk8: Readiness probe failed: dial tcp 10.42.0.14:50051: connect: connection refused (Unhealthy)","time":"2026-08-01T01:10:41+02:00"} +{"level":"warning","msg":"dev:threads Pod threads-5dcf65b76d-dwhk8 has critical status: Error. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T01:14:31+02:00"} +{"level":"warning","msg":"dev:threads Pod threads-5dcf65b76d-dwhk8 has critical status: CrashLoopBackOff. DevSpace will continue waiting, but this operation might timeout","time":"2026-08-01T01:14:42+02:00"} +{"level":"warning","msg":"dev:threads Pod threads-5dcf65b76d-dwhk8: Back-off restarting failed container threads in pod threads-5dcf65b76d-dwhk8_platform(86b305a0-80cb-40df-b6a1-2223d003d7bb) (BackOff)","time":"2026-08-01T01:14:52+02:00"} +{"level":"info","msg":"dev:threads Waiting for pod to become ready...","time":"2026-08-01T01:17:50+02:00"} +{"level":"warning","msg":"dev:threads Pod threads-5dcf65b76d-dwhk8: Back-off restarting failed container threads in pod threads-5dcf65b76d-dwhk8_platform(86b305a0-80cb-40df-b6a1-2223d003d7bb) (BackOff)","time":"2026-08-01T01:18:00+02:00"} +{"level":"warning","msg":"dev:threads Pod threads-5dcf65b76d-dwhk8: Readiness probe failed: dial tcp 10.42.0.14:50051: connect: connection refused (Unhealthy)","time":"2026-08-01T01:18:00+02:00"} +{"level":"info","msg":"dev:threads Waiting for pod to become ready...","time":"2026-08-01T01:19:08+02:00"} +{"level":"warning","msg":"dev:threads Pod threads-5b7cf54c7c-q4d7c: Readiness probe failed: dial tcp 10.42.0.15:50051: connect: connection refused (Unhealthy)","time":"2026-08-01T01:19:29+02:00"} +{"level":"info","msg":"dev:threads Waiting for pod to become ready...","time":"2026-08-01T01:25:25+02:00"} +{"level":"info","msg":"dev:threads Selected pod threads-68dd7fb865-8jsb5","time":"2026-08-01T01:25:26+02:00"} +{"level":"info","msg":"dev:threads ports Port forwarding started on: 50051 -\u003e 50051","time":"2026-08-01T01:25:27+02:00"} +{"level":"info","msg":"dev:threads sync Sync started on: ./ \u003c-\u003e /opt/app/data","time":"2026-08-01T01:25:29+02:00"} +{"level":"info","msg":"dev:threads sync Waiting for initial sync to complete","time":"2026-08-01T01:25:29+02:00"} +{"level":"info","msg":"dev:threads sync Initial sync completed","time":"2026-08-01T01:25:32+02:00"} +{"level":"info","msg":"dev:threads logs Streaming logs of pod:container threads-68dd7fb865-8jsb5:threads","time":"2026-08-01T01:25:32+02:00"} +{"level":"info","msg":"dev:threads logs threads: go: downloading github.com/jackc/pgx/v5 v5.8.0","time":"2026-08-01T01:25:34+02:00"} +{"level":"info","msg":"dev:threads logs threads: go: downloading google.golang.org/grpc v1.79.1","time":"2026-08-01T01:25:34+02:00"} +{"level":"info","msg":"dev:threads logs threads: go: downloading google.golang.org/protobuf v1.36.11","time":"2026-08-01T01:25:35+02:00"} +{"level":"info","msg":"dev:threads logs threads: go: downloading github.com/google/uuid v1.6.0","time":"2026-08-01T01:25:36+02:00"} +{"level":"info","msg":"dev:threads logs threads: go: downloading github.com/jackc/puddle/v2 v2.2.2","time":"2026-08-01T01:25:36+02:00"} +{"level":"info","msg":"dev:threads logs threads: go: downloading golang.org/x/net v0.48.0","time":"2026-08-01T01:25:36+02:00"} +{"level":"info","msg":"dev:threads logs threads: go: downloading google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217","time":"2026-08-01T01:25:36+02:00"} +{"level":"info","msg":"dev:threads logs threads: go: downloading github.com/jackc/pgpassfile v1.0.0","time":"2026-08-01T01:25:36+02:00"} +{"level":"info","msg":"dev:threads logs threads: go: downloading github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761","time":"2026-08-01T01:25:36+02:00"} +{"level":"info","msg":"dev:threads logs threads: go: downloading golang.org/x/text v0.32.0","time":"2026-08-01T01:25:36+02:00"} +{"level":"info","msg":"dev:threads logs threads: go: downloading golang.org/x/sync v0.19.0","time":"2026-08-01T01:25:36+02:00"} +{"level":"info","msg":"dev:threads logs threads: go: downloading golang.org/x/sys v0.39.0","time":"2026-08-01T01:25:36+02:00"} +{"level":"info","msg":"dev:threads logs threads: 2026/07/31 23:26:17 ThreadsService listening on :50051","time":"2026-08-01T01:26:17+02:00"} diff --git a/.devspace/logs/dev.threads.log b/.devspace/logs/dev.threads.log new file mode 100644 index 0000000..0ea6a88 --- /dev/null +++ b/.devspace/logs/dev.threads.log @@ -0,0 +1,7 @@ +{"level":"info","msg":"Inject devspacehelper...","time":"2026-08-01T01:25:27+02:00"} +{"level":"info","msg":"Successfully injected devspacehelper into pod platform/threads-68dd7fb865-8jsb5","time":"2026-08-01T01:25:29+02:00"} +{"level":"info","msg":"Start syncing","time":"2026-08-01T01:25:29+02:00"} +{"level":"info","msg":"Downstream - Initial sync completed","time":"2026-08-01T01:25:30+02:00"} +{"level":"info","msg":"Upstream - Upload 41 create change(s) (Uncompressed ~301.36 KB)","time":"2026-08-01T01:25:31+02:00"} +{"level":"info","msg":"Upstream - Successfully processed 41 change(s)","time":"2026-08-01T01:25:31+02:00"} +{"level":"info","msg":"Upstream - Initial sync completed","time":"2026-08-01T01:25:32+02:00"} diff --git a/.devspace/logs/errors.log b/.devspace/logs/errors.log new file mode 100644 index 0000000..f9f0534 --- /dev/null +++ b/.devspace/logs/errors.log @@ -0,0 +1,2 @@ +{"level":"error","msg":"Runtime error occurred: error closing listener: close tcp4 127.0.0.1:50051: use of closed network connection","time":"2026-08-01T01:26:17+02:00"} +{"level":"error","msg":"Runtime error occurred: error closing listener: close tcp6 [::1]:50051: use of closed network connection","time":"2026-08-01T01:26:17+02:00"} diff --git a/.devspace/logs/portforwarding.log b/.devspace/logs/portforwarding.log new file mode 100644 index 0000000..6949b75 --- /dev/null +++ b/.devspace/logs/portforwarding.log @@ -0,0 +1,2 @@ +Forwarding from 127.0.0.1:50051 -> 50051 +Forwarding from [::1]:50051 -> 50051 diff --git a/internal/server/default_thread_test.go b/internal/server/default_thread_test.go new file mode 100644 index 0000000..59e665d --- /dev/null +++ b/internal/server/default_thread_test.go @@ -0,0 +1,115 @@ +package server + +import ( + "context" + "strings" + "testing" + + agentsv1 "github.com/agynio/threads/.gen/go/agynio/api/agents/v1" + "github.com/google/uuid" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" +) + +func identityContext(identityID uuid.UUID, identityType string) context.Context { + return metadata.NewIncomingContext( + context.Background(), + metadata.Pairs(identityIDMetadataKey, identityID.String(), identityTypeMetadataKey, identityType), + ) +} + +func instanceWithDefaultThread(t *testing.T, instanceID uuid.UUID, defaultThreadID string) *stubAgentsService { + t.Helper() + instance := &agentsv1.AgentInstance{Meta: &agentsv1.EntityMeta{Id: instanceID.String()}} + if defaultThreadID != "" { + instance.DefaultThreadId = &defaultThreadID + } + return &stubAgentsService{ + t: t, + getInstanceFn: func(_ context.Context, req *agentsv1.GetInstanceRequest, _ ...grpc.CallOption) (*agentsv1.GetInstanceResponse, error) { + if req.GetId() != instanceID.String() { + t.Fatalf("expected the caller's own instance %s, got %s", instanceID, req.GetId()) + } + return &agentsv1.GetInstanceResponse{Instance: instance}, nil + }, + } +} + +// An instance leaves thread_id out and gets its default. Resolution is from the +// caller's own identity, so nothing the container carries can misroute it. +func TestResolveSendThreadUsesTheInstanceDefault(t *testing.T) { + instanceID := uuid.New() + defaultThreadID := uuid.New() + srv := New(nil, nil, nil, nil, instanceWithDefaultThread(t, instanceID, defaultThreadID.String()), nil) + + got, err := srv.resolveSendThread(identityContext(instanceID, agentInstanceIdentityType), "") + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if got != defaultThreadID { + t.Fatalf("expected %s, got %s", defaultThreadID, got) + } +} + +// A named thread is used as given; the instance is never consulted. This is +// what lets B reach a sub-thread rather than always answering its origin. +func TestResolveSendThreadPrefersAnExplicitThread(t *testing.T) { + instanceID := uuid.New() + explicitThreadID := uuid.New() + agents := &stubAgentsService{ + t: t, + getInstanceFn: func(context.Context, *agentsv1.GetInstanceRequest, ...grpc.CallOption) (*agentsv1.GetInstanceResponse, error) { + t.Fatal("expected no instance lookup when a thread is named") + return nil, nil + }, + } + srv := New(nil, nil, nil, nil, agents, nil) + + got, err := srv.resolveSendThread(identityContext(instanceID, agentInstanceIdentityType), explicitThreadID.String()) + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if got != explicitThreadID { + t.Fatalf("expected %s, got %s", explicitThreadID, got) + } +} + +// Only an instance has a default to fall back to. +func TestResolveSendThreadRejectsOmissionByOtherCallers(t *testing.T) { + for _, identityType := range []string{"user", "app", agentIdentityType} { + t.Run(identityType, func(t *testing.T) { + srv := New(nil, nil, nil, nil, &stubAgentsService{t: t}, nil) + + _, err := srv.resolveSendThread(identityContext(uuid.New(), identityType), "") + if status.Code(err) != codes.InvalidArgument { + t.Fatalf("expected InvalidArgument, got %v", err) + } + }) + } +} + +// "none" means the platform infers nothing, and an instance that was never +// given a thread has nowhere to send. Refused rather than guessed at. +func TestResolveSendThreadRejectsAnInstanceWithoutADefault(t *testing.T) { + instanceID := uuid.New() + srv := New(nil, nil, nil, nil, instanceWithDefaultThread(t, instanceID, ""), nil) + + _, err := srv.resolveSendThread(identityContext(instanceID, agentInstanceIdentityType), "") + if status.Code(err) != codes.FailedPrecondition { + t.Fatalf("expected FailedPrecondition, got %v", err) + } + if !strings.Contains(status.Convert(err).Message(), "no default thread") { + t.Fatalf("expected the reason to name the missing default, got %v", err) + } +} + +func TestResolveSendThreadRejectsAMalformedThread(t *testing.T) { + srv := New(nil, nil, nil, nil, &stubAgentsService{t: t}, nil) + + _, err := srv.resolveSendThread(identityContext(uuid.New(), agentInstanceIdentityType), "not-a-uuid") + if status.Code(err) != codes.InvalidArgument { + t.Fatalf("expected InvalidArgument, got %v", err) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index b195a7e..9ae81a0 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -47,7 +47,7 @@ const ( ) type threadStore interface { - CreateThread(ctx context.Context, organizationID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) + CreateThread(ctx context.Context, threadID uuid.UUID, organizationID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) ArchiveThread(ctx context.Context, threadID uuid.UUID) (store.Thread, error) DegradeThread(ctx context.Context, threadID uuid.UUID) (store.Thread, error) AddParticipant(ctx context.Context, threadID, participantID uuid.UUID, passive bool) (store.Thread, error) @@ -184,7 +184,11 @@ func (s *Server) CreateThread(ctx context.Context, req *threadsv1.CreateThreadRe participants = append(participants, store.ParticipantInput{ID: initiator.ID, Passive: false}) } participants = append(participants, resolved...) - finalParticipants, agentClassIDs, err := s.finalizeParticipants(ctx, participants, initiator.ID, hasInitiator) + // The id is settled here rather than in the store: creating an agent + // instance names the thread that is adding it, and that has to be the id + // the thread ends up with. + threadID := uuid.New() + finalParticipants, agentClassIDs, err := s.finalizeParticipants(ctx, participants, initiator.ID, hasInitiator, threadID) if err != nil { return nil, err } @@ -193,7 +197,7 @@ func (s *Server) CreateThread(ctx context.Context, req *threadsv1.CreateThreadRe return nil, err } - thread, err := s.store.CreateThread(ctx, organizationID, participants) + thread, err := s.store.CreateThread(ctx, threadID, organizationID, participants) if err != nil { return nil, toStatusError(err) } @@ -242,7 +246,9 @@ func (s *Server) requireCanInitiateAgentClasses(ctx context.Context, callerID uu return nil } -func (s *Server) finalizeParticipants(ctx context.Context, participants []store.ParticipantInput, initiatorID uuid.UUID, hasInitiator bool) ([]store.ParticipantInput, []uuid.UUID, error) { +// originThreadID is the thread the participants are being added to, reported +// to Agents so a class policy of origin can make it the instance default. +func (s *Server) finalizeParticipants(ctx context.Context, participants []store.ParticipantInput, initiatorID uuid.UUID, hasInitiator bool, originThreadID uuid.UUID) ([]store.ParticipantInput, []uuid.UUID, error) { if len(participants) == 0 { return nil, nil, nil } @@ -258,7 +264,7 @@ func (s *Server) finalizeParticipants(ctx context.Context, participants []store. seen := make(map[uuid.UUID]struct{}, len(participants)) agentClassIDs := make([]uuid.UUID, 0) for i, participant := range participants { - storedID, agentClassID, err := s.finalParticipantID(ctx, participant.ID, typesByID[participant.ID]) + storedID, agentClassID, err := s.finalParticipantID(ctx, participant.ID, typesByID[participant.ID], originThreadID) if err != nil { return nil, nil, err } @@ -323,10 +329,10 @@ func (s *Server) identityTypes(ctx context.Context, ids []uuid.UUID) (map[uuid.U return typesByID, nil } -func (s *Server) finalParticipantID(ctx context.Context, id uuid.UUID, identityType identityv1.IdentityType) (uuid.UUID, uuid.UUID, error) { +func (s *Server) finalParticipantID(ctx context.Context, id uuid.UUID, identityType identityv1.IdentityType, originThreadID uuid.UUID) (uuid.UUID, uuid.UUID, error) { switch identityType { case identityv1.IdentityType_IDENTITY_TYPE_AGENT: - instanceID, err := s.createAgentInstance(ctx, id) + instanceID, err := s.createAgentInstance(ctx, id, originThreadID) if err != nil { return uuid.UUID{}, uuid.UUID{}, err } @@ -347,11 +353,69 @@ func (s *Server) finalParticipantID(ctx context.Context, id uuid.UUID, identityT } } -func (s *Server) createAgentInstance(ctx context.Context, agentID uuid.UUID) (uuid.UUID, error) { +// resolveSendThread fills in a thread the caller left out. Only an agent +// instance may: it falls back to the thread it was created to serve, which is +// resolved here from the caller's own identity rather than from anything the +// container could carry, so a stale value cannot misroute a message. +// +// Deliberately not the thread of whatever woke the instance. In A -> B -> C, B +// is woken by C's reply on thread BC while owing its answer to A on thread AB; +// origin composes, the trigger does not. +func (s *Server) resolveSendThread(ctx context.Context, raw string) (uuid.UUID, error) { + if trimmed := strings.TrimSpace(raw); trimmed != "" { + threadID, err := parseUUID(trimmed) + if err != nil { + return uuid.UUID{}, status.Errorf(codes.InvalidArgument, "thread_id: %v", err) + } + return threadID, nil + } + identityID, err := identityIDFromContext(ctx) + if err != nil { + return uuid.UUID{}, err + } + if !strings.EqualFold(incomingMetadataValue(ctx, identityTypeMetadataKey), agentInstanceIdentityType) { + return uuid.UUID{}, status.Error(codes.InvalidArgument, "thread_id is required") + } if s.agents == nil { return uuid.UUID{}, status.Error(codes.Internal, "agents service not configured") } - response, err := s.agents.CreateInstance(ctx, &agentsv1.CreateInstanceRequest{AgentId: agentID.String()}) + response, err := s.agents.GetInstance(ctx, &agentsv1.GetInstanceRequest{Id: identityID.String()}) + if err != nil { + return uuid.UUID{}, status.Errorf(codes.Internal, "get agent instance: %v", err) + } + defaultThreadID := strings.TrimSpace(response.GetInstance().GetDefaultThreadId()) + if defaultThreadID == "" { + return uuid.UUID{}, status.Error(codes.FailedPrecondition, "thread_id is required: agent instance has no default thread") + } + threadID, err := parseUUID(defaultThreadID) + if err != nil { + return uuid.UUID{}, status.Errorf(codes.Internal, "agent instance default_thread_id: %v", err) + } + return threadID, nil +} + +func incomingMetadataValue(ctx context.Context, key string) string { + md, ok := metadata.FromIncomingContext(ctx) + if !ok { + return "" + } + return metadataValue(md, key) +} + +func protoString(value string) *string { return &value } + +// createAgentInstance reports the thread doing the adding as a creation +// circumstance. It is a fact, not an instruction: the class policy decides +// whether it becomes the instance's default thread. +func (s *Server) createAgentInstance(ctx context.Context, agentID uuid.UUID, originThreadID uuid.UUID) (uuid.UUID, error) { + if s.agents == nil { + return uuid.UUID{}, status.Error(codes.Internal, "agents service not configured") + } + request := &agentsv1.CreateInstanceRequest{AgentId: agentID.String()} + if originThreadID != uuid.Nil { + request.Context = &agentsv1.CreateInstanceContext{ThreadId: protoString(originThreadID.String())} + } + response, err := s.agents.CreateInstance(ctx, request) if err != nil { return uuid.UUID{}, status.Errorf(codes.Internal, "create agent instance: %v", err) } @@ -459,7 +523,7 @@ func (s *Server) AddParticipant(ctx context.Context, req *threadsv1.AddParticipa if err != nil { return nil, err } - finalParticipants, agentClassIDs, err := s.finalizeParticipants(ctx, []store.ParticipantInput{{ID: participantID, Passive: false}}, uuid.Nil, false) + finalParticipants, agentClassIDs, err := s.finalizeParticipants(ctx, []store.ParticipantInput{{ID: participantID, Passive: false}}, uuid.Nil, false, threadID) if err != nil { return nil, err } @@ -484,9 +548,9 @@ func (s *Server) AddParticipant(ctx context.Context, req *threadsv1.AddParticipa } func (s *Server) SendMessage(ctx context.Context, req *threadsv1.SendMessageRequest) (*threadsv1.SendMessageResponse, error) { - threadID, err := parseUUID(req.GetThreadId()) + threadID, err := s.resolveSendThread(ctx, req.GetThreadId()) if err != nil { - return nil, status.Errorf(codes.InvalidArgument, "thread_id: %v", err) + return nil, err } if req.GetBody() == "" && len(req.GetFileIds()) == 0 { return nil, status.Error(codes.InvalidArgument, "body or file_ids must be provided") diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 0fa0520..281b096 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -24,7 +24,7 @@ import ( type stubThreadStore struct { t *testing.T - createThreadFn func(ctx context.Context, organizationID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) + createThreadFn func(ctx context.Context, threadID uuid.UUID, organizationID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) archiveThreadFn func(ctx context.Context, threadID uuid.UUID) (store.Thread, error) degradeThreadFn func(ctx context.Context, threadID uuid.UUID) (store.Thread, error) addParticipantFn func(ctx context.Context, threadID, participantID uuid.UUID, passive bool) (store.Thread, error) @@ -45,12 +45,12 @@ func (s *stubThreadStore) unexpectedCall(method string) { s.t.Fatalf("unexpected %s call", method) } -func (s *stubThreadStore) CreateThread(ctx context.Context, organizationID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { +func (s *stubThreadStore) CreateThread(ctx context.Context, threadID uuid.UUID, organizationID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { s.t.Helper() if s.createThreadFn == nil { s.t.Fatalf("unexpected CreateThread call") } - return s.createThreadFn(ctx, organizationID, participants) + return s.createThreadFn(ctx, threadID, organizationID, participants) } func (s *stubThreadStore) ArchiveThread(ctx context.Context, threadID uuid.UUID) (store.Thread, error) { @@ -314,7 +314,7 @@ func TestCreateThreadRecordsUsageWithCreatedThreadOrganization(t *testing.T) { storeStub := &stubThreadStore{ t: t, - createThreadFn: func(ctx context.Context, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { + createThreadFn: func(ctx context.Context, _ uuid.UUID, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { if orgID != organizationID { t.Fatalf("expected organization %s, got %s", organizationID, orgID) } @@ -402,7 +402,7 @@ func TestCreateThreadAgentInitiatorPassive(t *testing.T) { storeStub := &stubThreadStore{ t: t, - createThreadFn: func(ctx context.Context, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { + createThreadFn: func(ctx context.Context, _ uuid.UUID, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { storeCalled = true if orgID != organizationID { t.Fatalf("expected organization %s, got %s", organizationID, orgID) @@ -472,7 +472,7 @@ func TestCreateThreadEmptyParticipantsWithAgentInitiator(t *testing.T) { storeStub := &stubThreadStore{ t: t, - createThreadFn: func(ctx context.Context, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { + createThreadFn: func(ctx context.Context, _ uuid.UUID, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { storeCalled = true if orgID != organizationID { t.Fatalf("expected organization %s, got %s", organizationID, orgID) @@ -532,7 +532,7 @@ func TestCreateThreadUserInitiatorActive(t *testing.T) { storeStub := &stubThreadStore{ t: t, - createThreadFn: func(ctx context.Context, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { + createThreadFn: func(ctx context.Context, _ uuid.UUID, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { storeCalled = true if orgID != organizationID { t.Fatalf("expected organization %s, got %s", organizationID, orgID) @@ -596,7 +596,7 @@ func TestCreateThreadMissingIdentityMetadataRejected(t *testing.T) { storeStub := &stubThreadStore{ t: t, - createThreadFn: func(ctx context.Context, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { + createThreadFn: func(ctx context.Context, _ uuid.UUID, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { storeCalled = true return store.Thread{}, nil }, @@ -661,7 +661,7 @@ func TestCreateThreadNicknameUsesOrganizationID(t *testing.T) { storeStub := &stubThreadStore{ t: t, - createThreadFn: func(ctx context.Context, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { + createThreadFn: func(ctx context.Context, _ uuid.UUID, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { storeCalled = true if orgID != organizationID { t.Fatalf("expected organization %s, got %s", organizationID, orgID) @@ -751,7 +751,7 @@ func TestCreateThreadNicknameUsesOrganizationIDFromAgentIdentity(t *testing.T) { storeStub := &stubThreadStore{ t: t, - createThreadFn: func(ctx context.Context, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { + createThreadFn: func(ctx context.Context, _ uuid.UUID, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { storeCalled = true if orgID != organizationID { t.Fatalf("expected organization %s, got %s", organizationID, orgID) @@ -842,7 +842,7 @@ func TestCreateThreadMixedParticipantIdentifiers(t *testing.T) { storeStub := &stubThreadStore{ t: t, - createThreadFn: func(ctx context.Context, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { + createThreadFn: func(ctx context.Context, _ uuid.UUID, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { storeCalled = true if orgID != organizationID { t.Fatalf("expected organization %s, got %s", organizationID, orgID) @@ -1007,7 +1007,7 @@ func assertCreateThreadDedupesInitiator(t *testing.T, initiatorID, participantID storeStub := &stubThreadStore{ t: t, - createThreadFn: func(ctx context.Context, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { + createThreadFn: func(ctx context.Context, _ uuid.UUID, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { storeCalled = true if orgID != organizationID { t.Fatalf("expected organization %s, got %s", organizationID, orgID) @@ -1062,7 +1062,7 @@ func TestCreateThreadAuthorizationDenied(t *testing.T) { storeStub := &stubThreadStore{ t: t, - createThreadFn: func(ctx context.Context, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { + createThreadFn: func(ctx context.Context, _ uuid.UUID, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { storeCalled = true return store.Thread{}, nil }, @@ -1125,7 +1125,7 @@ func TestCreateThreadWritesAuthorizationTuples(t *testing.T) { storeStub := &stubThreadStore{ t: t, - createThreadFn: func(ctx context.Context, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { + createThreadFn: func(ctx context.Context, _ uuid.UUID, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { storeCalled = true if orgID != organizationID { t.Fatalf("expected organization %s, got %s", organizationID, orgID) @@ -2860,7 +2860,7 @@ func TestCreateThreadStoresAgentInstanceForAgentClassParticipant(t *testing.T) { storeStub := &stubThreadStore{ t: t, - createThreadFn: func(ctx context.Context, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { + createThreadFn: func(ctx context.Context, _ uuid.UUID, orgID uuid.UUID, participants []store.ParticipantInput) (store.Thread, error) { if orgID != organizationID { t.Fatalf("expected organization %s, got %s", organizationID, orgID) } diff --git a/internal/store/threads.go b/internal/store/threads.go index 000a8ce..90d49da 100644 --- a/internal/store/threads.go +++ b/internal/store/threads.go @@ -12,10 +12,13 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) -func (s *Store) CreateThread(ctx context.Context, organizationID uuid.UUID, participantInputs []ParticipantInput) (Thread, error) { +// CreateThread takes the thread's id rather than inventing one, because +// participants are finalized before the row exists: creating an agent instance +// has to name the thread that is adding it, and that has to be the same id the +// thread ends up with. +func (s *Store) CreateThread(ctx context.Context, threadID uuid.UUID, organizationID uuid.UUID, participantInputs []ParticipantInput) (Thread, error) { var thread Thread err := s.runTx(ctx, func(tx pgx.Tx) error { - threadID := uuid.New() now := time.Now().UTC() if _, err := tx.Exec(ctx, `INSERT INTO threads (id, organization_id, status, created_at, updated_at, message_count) VALUES ($1, $2, $3, $4, $4, 0)`, threadID, organizationID, int16(ThreadStatusActive), now); err != nil { return err