Files
multica/server/internal/handler/daemon_task_lookup_test.go
13eb4bda93 MUL-8154 fix(daemon): enforce runtime ownership on daemon routes (#9423)
* fix(daemon): enforce runtime ownership on daemon routes (MUL-8154)

/api/daemon/* accepts any workspace member's PAT/JWT, and authorization only
checked workspace membership. Any member could claim another member's tasks
(receiving the agent's custom_env, MCP config and an owner-scoped task token),
write task results/messages, and take a runtime over by re-registering its
daemon_id.

- Runtime-level check on every runtime/task daemon route, HTTP heartbeat,
  deregister, batch claim (and so the WS claim RPC) and the WS handshake:
  user credentials must own the runtime (NULL owner fails closed); daemon
  tokens must match a non-empty runtime daemon_id.
- Register: pre-check that every existing row on the daemon_id belongs to the
  caller, plus a required_owner_id fence in both upserts (incl. the
  failed-profile path) so a concurrent registration cannot win the row.
  Daemon-token registration keeps the existing owner.
- Legacy merge: only for user-credential registrations, and only when the
  caller owns both rows, verified on the FOR UPDATE-locked rows.
- Tasks without a runtime are not writable through daemon routes; status
  stays readable.
- GetTaskStatus / WS lease queries carry runtime owner/daemon so the hot
  paths keep one round trip.

Co-authored-by: multica-agent <[email protected]>

* test(daemon): authenticate fixtures as the runtime owner (MUL-8154)

Most handler tests drove daemon routes with a daemon token whose daemon_id
did not match the fixture runtime, relying on the workspace-wide token scope
this fix removes. They now authenticate the way real daemons do, with the
runtime owner's user credential (newOwnerDaemonRequest). Tests about daemon
token scoping keep mdt_ and get fixtures with a matching daemon_id; the
ownerless-runtime tests also assert that a user credential is refused.

Co-authored-by: multica-agent <[email protected]>

* fix(daemon): serialize registration per daemon and keep Qoder WS claims working (MUL-8154)

Review of #9423:

- P1: the owner pre-check and the per-provider/profile upserts hit different
  conflict keys, so two members registering different providers of the same
  daemon could both pass the pre-check and both insert, leaving one daemon
  with two owners (and a daemon token for either runtime reaching the
  other's). Registration now runs the pre-check and every runtime write in one
  transaction under a pg_advisory_xact_lock on (workspace_id, daemon_id), on a
  single connection so queued registrations cannot starve the holder. Legacy
  merges, analytics and the register event run after commit; failed-profile
  records use savepoints so they stay best-effort.
- P2: the Qoder revocation check read the Authorization header, which the WS
  RPC bridge's in-process request does not carry, so valid Qoder WS claims
  silently returned nothing. DaemonAuth now records the mdt_ token hash in the
  request context, the WS identity keeps it, the RPC bridge re-attaches it,
  and the check validates that hash on every call (revocation still applies
  to an open connection).
- Regression tests for cross-provider and provider+profile concurrent
  registration, and for valid/revoked Qoder credentials over WS RPC.

Co-authored-by: multica-agent <[email protected]>

* fix(daemon): bind daemon tokens to the runtime owner they were issued for (MUL-8154)

Second review of #9423: a daemon_id can be re-registered by another member
after its runtimes are deleted, and a daemon token was bound only to
(workspace, daemon_id), so the previous owner's Remote MCP token reached the
new owner's runtime (heartbeat, claim, owner-scoped task token).

- daemon_token.owner_id (migration 588). Remote MCP tokens are bound to the
  locked runtime owner inside FinalizeTaskClaim; live tokens are backfilled
  only where their daemon has exactly one owner.
- DaemonAuth carries the owner binding (DB and cache; cache key bumped to v2
  so pre-upgrade entries are never read), the WS identity keeps it and the
  RPC bridge replays it.
- Daemon-token authorization now also requires token owner == runtime owner.
  Qoder-managed tokens stay unbound (governed by cloud_agent_connection), and
  user credentials can no longer register the reserved qoder-managed-* ids.
- Tests: the reuse lifecycle over HTTP, a new WS handshake and an already
  open WS connection; the reserved daemon_id; the concurrency barrier now
  hooks the registration transaction (TxStarter) so the interleaving is real.

Co-authored-by: multica-agent <[email protected]>

* fix(daemon): owner-bind daemon-token registration; drop unsafe token backfill (MUL-8154)

Third review of #9423:

- Migration 588 no longer backfills daemon_token.owner_id from the daemon's
  current owner: that owner may be a later re-registrant, which re-bound the
  previous owner's live token to them. Existing non-Qoder tokens stay NULL and
  are refused (24h Remote MCP credentials, re-issued on the next claim).
- owner_id is a plain UUID column, no foreign key / cascade (AGENTS.md).
- Register via a non-Qoder daemon token now uses the token's owner for the
  in-transaction pre-check and the SQL owner fence, and refuses unbound
  tokens. Qoder keeps its connection-based check; daemon tokens still never
  trigger legacy merges.
- Tests: an old owner's bound token and an unbound legacy token can neither
  register over nor heartbeat the replacement owner's runtime.

Co-authored-by: multica-agent <[email protected]>

* fix(daemon): grant the Qoder registration exemption only with a live connection (MUL-8154)

Fourth review of #9423: register skipped the token owner binding for any
qoder-managed-* daemon token, but checked the Qoder connection only when the
body declared a qoder_cloud runtime. A pre-upgrade Remote MCP token under that
prefix (no owner, no connection) could register type=claude and rewrite
another member's runtime.

Register now verifies the token against a live cloud_agent_connection before
anything else whenever the token is in the Qoder namespace, regardless of the
declared runtimes, and restricts that namespace to the bridge's own shape: a
built-in qoder_cloud runtime, no custom profiles, no failed_profiles.

Test covers built-in, custom-profile, failed-profiles-only and qoder_cloud
bodies with an unconnected token, plus the connected token's allowed and
refused shapes.

Co-authored-by: multica-agent <[email protected]>

---------

Co-authored-by: multica-eve <[email protected]>
Co-authored-by: multica-agent <[email protected]>
2026-10-22 16:47:06 +08:00

191 lines
8.0 KiB
Go

package handler
import (
"context"
"net/http"
"strings"
"testing"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/multica-ai/multica/server/internal/service"
"github.com/multica-ai/multica/server/internal/testutil"
db "github.com/multica-ai/multica/server/pkg/db/generated"
)
// The daemon interrupts a running agent the moment a task-status poll answers
// `404 task not found` (shouldInterruptAgent → isTaskNotFoundError). So that
// body is a kill signal, and only a lookup that completed and found no task row
// may produce it. Transient failures and authorization misses must stay distinct.
// lookupFaultPool fails one named query and passes everything else through, so
// a single link in ResolveTaskWorkspaceIDChecked can be made to time out while
// the task row itself stays readable.
type lookupFaultPool struct {
db.DBTX
query string
called bool
}
func (f *lookupFaultPool) QueryRow(ctx context.Context, query string, args ...any) pgx.Row {
if strings.Contains(query, "-- name: "+f.query+" :one") {
f.called = true
return &mockRow{err: context.DeadlineExceeded}
}
return f.DBTX.QueryRow(ctx, query, args...)
}
// TestGetTaskStatus_DoesNotResolveSourceWorkspace pins the hot-path contract:
// status polling authorizes through the owning agent's workspace and never
// follows optional issue / chat / autopilot links.
func TestGetTaskStatus_DoesNotResolveSourceWorkspace(t *testing.T) {
runtimeID := dbfx.Runtime(t, "MUL-7259 lookup runtime")
agentID := dbfx.Agent(t, "MUL-7259 lookup agent", runtimeID)
issueID := dbfx.Issue(t, "MUL-7259 lookup issue")
chatID := dbfx.ChatSession(t, agentID)
apID := dbfx.Insert(t, "autopilot", testutil.Cols{
"workspace_id": testWorkspaceID, "title": "MUL-7259 lookup autopilot",
"assignee_id": agentID, "status": "paused", "created_by_type": "member", "created_by_id": testUserID,
})
runID := dbfx.Insert(t, "autopilot_run", testutil.Cols{"autopilot_id": apID, "source": "manual", "status": "running"})
for _, tc := range []struct{ query, column, id string }{
{"GetIssue", "issue_id", issueID},
{"GetChatSession", "chat_session_id", chatID},
{"GetAutopilotRun", "autopilot_run_id", runID},
{"GetAutopilot", "autopilot_run_id", runID},
} {
t.Run(tc.query, func(t *testing.T) {
taskID := dbfx.Task(t, agentID, testutil.Cols{
"runtime_id": runtimeID, "status": "running", "started_at": testutil.Raw("now()"), tc.column: tc.id,
})
fault := &lookupFaultPool{DBTX: testPool, query: tc.query}
h := &Handler{Queries: db.New(fault), TaskService: &service.TaskService{Queries: db.New(fault)}}
req := newOwnerDaemonRequest(http.MethodGet, "/api/daemon/tasks/"+taskID+"/status", nil)
req = withURLParam(req, "taskId", taskID)
var response map[string]string
testutil.Call(t, h.GetTaskStatus, req).Want(http.StatusOK).JSON(&response)
if fault.called {
t.Fatalf("status poll unexpectedly executed %s", tc.query)
}
if response["status"] != "running" {
t.Fatalf("status = %q, want running", response["status"])
}
})
}
}
// TestGetTaskStatus_TaskRowPresenceContract distinguishes a genuinely missing
// task from a surviving task whose optional source row has gone away.
func TestGetTaskStatus_TaskRowPresenceContract(t *testing.T) {
runtimeID := dbfx.Runtime(t, "MUL-7259 absent runtime")
agentID := dbfx.Agent(t, "MUL-7259 absent agent", runtimeID)
t.Run("task row missing", func(t *testing.T) {
missing := uuid.NewString()
req := newOwnerDaemonRequest(http.MethodGet, "/api/daemon/tasks/"+missing+"/status", nil)
req = withURLParam(req, "taskId", missing)
w := testutil.Call(t, testHandler.GetTaskStatus, req).Want(http.StatusNotFound)
if !strings.Contains(w.Body.String(), "task not found") {
t.Fatalf("a genuinely missing task must still interrupt the daemon: %s", w.Body.String())
}
})
t.Run("optional source missing", func(t *testing.T) {
// agent_task_queue.issue_id is ON DELETE CASCADE, so an issue task can
// never outlive its issue. chat_session_id is ON DELETE SET NULL, so a
// chat task can survive its source. The status endpoint now authorizes
// that row through its owning agent instead of treating the optional
// source as task identity.
chatID := dbfx.ChatSession(t, agentID)
taskID := dbfx.Task(t, agentID, testutil.Cols{
"runtime_id": runtimeID, "status": "running",
"started_at": testutil.Raw("now()"), "chat_session_id": chatID,
})
dbfx.Exec(t, "DELETE FROM chat_session WHERE id = $1", chatID)
req := newOwnerDaemonRequest(http.MethodGet, "/api/daemon/tasks/"+taskID+"/status", nil)
req = withURLParam(req, "taskId", taskID)
var response map[string]string
testutil.Call(t, testHandler.GetTaskStatus, req).Want(http.StatusOK).JSON(&response)
if response["status"] != "running" {
t.Fatalf("status = %q, want running", response["status"])
}
})
}
// TestGetTaskStatus_ForeignWorkspace_Returns404 pins the permission boundary:
// splitting lookup failures out of the 404 must not turn a cross-workspace task
// into a distinguishable response. A foreign task and a missing one look alike.
func TestGetTaskStatus_ForeignWorkspace_Returns404(t *testing.T) {
runtimeID := dbfx.Runtime(t, "MUL-7259 foreign runtime")
agentID := dbfx.Agent(t, "MUL-7259 foreign agent", runtimeID)
issueID := dbfx.Issue(t, "MUL-7259 foreign issue")
taskID := dbfx.Task(t, agentID, testutil.Cols{
"runtime_id": runtimeID, "issue_id": issueID,
"status": "running", "started_at": testutil.Raw("now()"),
})
otherWorkspace := uuid.NewString()
req := newDaemonTokenRequest(http.MethodGet, "/api/daemon/tasks/"+taskID+"/status", nil, otherWorkspace, "other-daemon")
req = withURLParam(req, "taskId", taskID)
w := testutil.Call(t, testHandler.GetTaskStatus, req).Want(http.StatusNotFound)
if strings.Contains(w.Body.String(), "task not found") {
t.Fatalf("authorization miss must not carry the daemon's deletion signal: %s", w.Body.String())
}
}
// TestResolveTaskWorkspaceIDChecked_SeparatesAbsenceFromFailure asserts the
// distinction at the service boundary, where the two callers now rely on it.
func TestResolveTaskWorkspaceIDChecked_SeparatesAbsenceFromFailure(t *testing.T) {
ctx := context.Background()
runtimeID := dbfx.Runtime(t, "MUL-7259 resolver runtime")
agentID := dbfx.Agent(t, "MUL-7259 resolver agent", runtimeID)
issueID := dbfx.Issue(t, "MUL-7259 resolver issue")
t.Run("resolves", func(t *testing.T) {
taskID := dbfx.Task(t, agentID, testutil.Cols{"runtime_id": runtimeID, "issue_id": issueID})
task, err := testHandler.Queries.GetAgentTask(ctx, parseUUID(taskID))
if err != nil {
t.Fatal(err)
}
got, err := testHandler.TaskService.ResolveTaskWorkspaceIDChecked(ctx, task)
if err != nil || got != testWorkspaceID {
t.Fatalf("workspace=%q err=%v, want %q / nil", got, err, testWorkspaceID)
}
})
t.Run("absent link is not an error", func(t *testing.T) {
chatID := dbfx.ChatSession(t, agentID)
taskID := dbfx.Task(t, agentID, testutil.Cols{"runtime_id": runtimeID, "chat_session_id": chatID})
dbfx.Exec(t, "DELETE FROM chat_session WHERE id = $1", chatID)
task, err := testHandler.Queries.GetAgentTask(ctx, parseUUID(taskID))
if err != nil {
t.Fatal(err)
}
got, err := testHandler.TaskService.ResolveTaskWorkspaceIDChecked(ctx, task)
if got != "" || err != nil {
t.Fatalf("workspace=%q err=%v, want \"\" / nil", got, err)
}
})
t.Run("failed lookup is an error", func(t *testing.T) {
taskID := dbfx.Task(t, agentID, testutil.Cols{"runtime_id": runtimeID, "issue_id": issueID})
task, err := testHandler.Queries.GetAgentTask(ctx, parseUUID(taskID))
if err != nil {
t.Fatal(err)
}
svc := &service.TaskService{Queries: db.New(&lookupFaultPool{DBTX: testPool, query: "GetIssue"})}
got, err := svc.ResolveTaskWorkspaceIDChecked(ctx, task)
if got != "" || err == nil {
t.Fatalf("workspace=%q err=%v, want \"\" / error", got, err)
}
// The best-effort wrapper keeps flattening both cases to "" — broadcast
// callers depend on that and must not start panicking on a blip.
if ws := svc.ResolveTaskWorkspaceID(ctx, task); ws != "" {
t.Fatalf("best-effort resolver should still return \"\", got %q", ws)
}
})
}