fix(gateway): stop run.subscribe from leaking another workspace's run events (#2961)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

apps/backend/AGENTS.md

@@ -177,7 +177,7 @@ Standalone agentctl is launched in its own process group so terminal Ctrl+C is h
 - **New user-facing service entry points must apply scoping** — call the `authorize*` helpers in `task/service/service_access.go` (or the same pattern) when adding routes that read or mutate workspace-scoped data. This includes **session-keyed** entry points: any service method taking a caller-supplied `sessionID` (or a `taskID` used to reach sessions) authorizes via `authorizeTaskID(session.TaskID)`. Authorize off the row you already read rather than calling `AuthorizeSessionAccess`, which re-reads the session. Batch helpers that take ID *lists* (`BatchGetSessionsForTasks`, `GetPrimarySessionInfoForTasks`, …) are exempt by convention: their callers derive the IDs from an already-authorized task list, so per-ID checks would add N queries to list views for no gain — keep it that way, and don't hand them caller-supplied IDs.
 - **Two session paths bypass the task service and must guard themselves.** (1) Handlers that resolve an execution by a bare in-memory lookup (`GetExecutionBySessionID`, `*BySessionID`) skip the `GetOrEnsure*` chokepoint where the lifecycle check runs — call `service.AuthorizeSessionAccess` (see `ProcessHandlers.denySessionAccess`) or `lifecycleMgr.CheckSessionAccess` first. (2) The orchestrator reads sessions through its own repo handle, so it inherits nothing from the task service; session-keyed entry points there call `authorizeSession` / `authorizeTask`, wired by `SetSessionAccessChecker` / `SetTaskAccessChecker`. An entry point that accepts **both** a task and a session ID must use `authorizeTaskSessionPair`, never `authorizeSession` alone: the caller can satisfy a session-only check with one of their own sessions while pointing `taskID` at someone else's task, which the method then uses for its task-scoped work. Put the guard **first** in the method, before any repo/executor/agent-manager use — `TestSessionKeyedEntryPointsGuardBeforeDependencies` runs each entry point with nil dependencies and fails on a panic if a guard is placed too late.
 - **Dispatched WS actions have a gateway backstop.** `Client.authorizeAction` (`internal/gateway/websocket/dispatch_scope.go`) checks any payload carrying `task_id` or `session_id` against the client's identity *before* dispatch, so a newly added action is scoped by default rather than only when its author remembers. Handler-level scoping is still required (defense in depth, and it covers non-WS callers) — do not delete a service-layer check because the backstop exists. The backstop reads `task_id`, `session_id` and `task_environment_id`, plus `id` on **top-level `task.<verb>` actions only** (`task.state`, `task.move`, `task.get`, … — `task.` and exactly one more segment). Keep using those names for task-scoped actions, because an action that invents a new name for the same kind of resource silently opts out. (`user_shell.stop` was exactly that: it keys off `task_environment_id` with `task_id` optional, so a `task_id`-only backstop missed it. `task.state` and `task.move` were the second instance: they name the task `id`, so the backstop parsed no refs and let one user mutate any other user's task — hence the `id` rule.) Deeper `task.*` namespaces are deliberately excluded: in `task.plan.revision.get` and `task.review.finding.update` the `id` names a revision or a finding, so those actions must carry `task_id` when they are task-scoped. Also avoid permissive types (`any`, `json.RawMessage`) for those fields: a payload that fails to unmarshal into the ref struct is treated as naming nothing.
-- **WS**: clients carry their identity; dispatched actions and subscriptions are scoped; workspace-carrying events route via `Hub.BroadcastToWorkspace`. A new `hub.Broadcast` (global) call site needs a `//ws:global` justification comment.
+- **WS**: clients carry their identity; dispatched actions and subscriptions are scoped; workspace-carrying events route via `Hub.BroadcastToWorkspace`. A new `hub.Broadcast` (global) call site needs a `//ws:global` justification comment. Subscription actions (`*.subscribe`) `return` before reaching the dispatch backstop (`Client.authorizeAction`), so scoping a new subscribe topic means adding a hook to `SubscriptionAccessPolicy` (`internal/gateway/websocket/access.go`) *and* calling it explicitly inside that action's own handler — the backstop's `scopedActionRefs` cannot reach it. `run.subscribe`'s `Subscriptions.Run` hook (added for WO-02, resolving the run's workspace via `runs.GetRunWorkspaceID`) is the worked example.
 - **Task overview vs. session detail:** Keep cross-task state in the bounded, persisted `TaskStatusSummary` projection and `task.status_summary.updated`; keep unbounded transcript and execution data session-scoped. The [bounded task-status spec](../../docs/specs/platform/bounded-task-status-delivery.md) and [accepted ADR](../../docs/decisions/2026-08-01-separate-task-summary-session-stream-traffic.md) define the shared contract.
 - **Port-proxy subtree capability:** with auth enforced, `/port-proxy/:sessionId/:port/*` needs a credential on *every* request, but browser subresource fetches cannot always carry one (the `<link rel="manifest">` fetch never sends cookies; sandboxed iframes and `?token=`-opened previews drop them elsewhere). `PortProxyHandler` mints a short-lived (15 min, sliding), path-scoped, HMAC-signed capability after the document authenticates and propagates it as a cookie scoped to the proxy subtree plus a `kandev_cap` query parameter appended to rewritten asset URLs. `requireConnectionAuth` (`internal/gateway/websocket/access.go`) accepts either form only for the exact session:port it was minted for and restores the issuing identity, so `CheckSessionAccess` still gates on the real owner. Synthetic identities (auth disabled) skip minting — proxied bodies stay byte-identical to the pre-auth behavior.
 - **Self-authenticating webhooks** (automation and office channels) and `/health`, `/api/v1/features`, `/api/v1/app-state` stay public — the allowlist lives in `auth/httpmw/middleware.go` with a pinning test. Plugin webhooks are not allowlisted: `auth/httpmw` structurally defers only GET/POST relay paths because it cannot read manifests, and `internal/plugins.Controller.webhook` enforces each manifest's `access` policy.

apps/backend/internal/backendapp/auth.go

@@ -14,6 +14,8 @@ import (
 	"github.com/kandev/kandev/internal/common/logger"
 	"github.com/kandev/kandev/internal/db"
 	gateways "github.com/kandev/kandev/internal/gateway/websocket"
+	officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite"
+	"github.com/kandev/kandev/internal/task/repository/repoerrors"
 	sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
 	taskservice "github.com/kandev/kandev/internal/task/service"
 )
@@ -84,7 +86,9 @@ func provideAuthService(
 // gatewayAuthPolicy assembles the WS gateway scoping hooks from the auth and
 // task services. The workspace-owner resolver caches lookups briefly: it runs
 // on every workspace-scoped broadcast.
-func gatewayAuthPolicy(authSvc *auth.Service, taskSvc *taskservice.Service, taskRepo *sqliterepo.Repository) gateways.AuthPolicy {
+func gatewayAuthPolicy(
+	authSvc *auth.Service, taskSvc *taskservice.Service, taskRepo *sqliterepo.Repository, officeRepo *officesqlite.Repository,
+) gateways.AuthPolicy {
 	return gateways.AuthPolicy{
 		Enforced:     func() bool { return authSvc.Mode() != auth.ModeDisabled },
 		ResolveToken: authSvc.ResolveBearer,
@@ -98,6 +102,7 @@ func gatewayAuthPolicy(authSvc *auth.Service, taskSvc *taskservice.Service, task
 		Subscriptions: gateways.SubscriptionAccessPolicy{
 			Task:    taskSvc.AuthorizeTaskAccess,
 			Session: taskSvc.AuthorizeSessionAccess,
+			Run:     runSubscriptionCheck(taskSvc, officeRepo),
 		},
 		WorkspaceOwner: newWorkspaceOwnerResolver(taskRepo),
 		// The user-shell actions name a task environment and treat task_id as
@@ -106,6 +111,38 @@ func gatewayAuthPolicy(authSvc *auth.Service, taskSvc *taskservice.Service, task
 	}
 }
 
+// runSubscriptionCheck resolves a run's owning workspace (via its agent
+// profile) and defers to the task service's workspace visibility rule.
+// Unlike WorkspaceOwner (which runs on every workspace broadcast), this
+// runs once per run.subscribe — a rare control message — so it is not
+// cached.
+//
+// An ordinary (non-Office) Kanban agent profile has workspace_id=""
+// (schema default, never backfilled) — the dominant case, since the run
+// processor and workflow-engine queue_run dispatch run for every backend,
+// not just Office. AuthorizeWorkspaceAccess's workspaceID=="" branch means
+// "no workspace scoping applies" (used for dangling references elsewhere
+// in the task service), which would silently allow subscribing to these
+// runs; that meaning does not apply here, so an empty resolution is denied
+// outright before it ever reaches that helper. officeRepo==nil fails
+// closed for the same reason: this is a security check, not a visibility
+// fallback.
+func runSubscriptionCheck(taskSvc *taskservice.Service, officeRepo *officesqlite.Repository) func(context.Context, string) error {
+	return func(ctx context.Context, runID string) error {
+		if officeRepo == nil {
+			return repoerrors.ErrWorkspaceNotFound
+		}
+		workspaceID, err := officeRepo.GetRunWorkspaceID(ctx, runID)
+		if err != nil {
+			return err
+		}
+		if workspaceID == "" {
+			return repoerrors.ErrWorkspaceNotFound
+		}
+		return taskSvc.AuthorizeWorkspaceAccess(ctx, workspaceID)
+	}
+}
+
 // ownerCacheTTL bounds staleness of broadcast routing after an ownership
 // change (only the setup wizard's claim changes owners in practice).
 const ownerCacheTTL = 30 * time.Second

apps/backend/internal/backendapp/auth_test.go

@@ -4,15 +4,27 @@ import (
 	"context"
 	"net/http"
 	"net/http/httptest"
+	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/jmoiron/sqlx"
 	_ "github.com/mattn/go-sqlite3"
 
+	settingsstore "github.com/kandev/kandev/internal/agent/settings/store"
 	"github.com/kandev/kandev/internal/auth"
+	"github.com/kandev/kandev/internal/auth/authn"
 	authstore "github.com/kandev/kandev/internal/auth/store"
 	"github.com/kandev/kandev/internal/common/config"
+	"github.com/kandev/kandev/internal/common/logger"
+	"github.com/kandev/kandev/internal/db"
+	"github.com/kandev/kandev/internal/events/bus"
+	"github.com/kandev/kandev/internal/office/models"
+	officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite"
+	"github.com/kandev/kandev/internal/task/repository"
+	sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
+	taskservice "github.com/kandev/kandev/internal/task/service"
 	userstore "github.com/kandev/kandev/internal/user/store"
 )
 
@@ -116,3 +128,206 @@ func TestPluginSSOBridgeDefaultPortPlainName(t *testing.T) {
 		t.Fatalf("SSO Set-Cookie names = %v, want exactly [kandev_session]", names)
 	}
 }
+
+// newRunSubscriptionCheckHarness builds a task service, task repository, and
+// office repository over one shared SQLite database, the same pairing
+// gatewayAuthPolicy wires in production. WO-02 round 2's runSubscriptionCheck
+// tests seed agent profiles and runs against it; round 3 also seeds owned
+// workspaces via the returned task repository to drive real ownership
+// decisions instead of the unscoped-caller shortcut.
+func newRunSubscriptionCheckHarness(t *testing.T) (*taskservice.Service, *sqliterepo.Repository, *officesqlite.Repository) {
+	t.Helper()
+	dbConn, err := db.OpenSQLite(filepath.Join(t.TempDir(), "run-subscribe.db"))
+	if err != nil {
+		t.Fatalf("open sqlite: %v", err)
+	}
+	database := sqlx.NewDb(dbConn, "sqlite3")
+	t.Cleanup(func() { _ = database.Close() })
+	taskRepo, cleanup, err := repository.Provide(database, database, nil)
+	if err != nil {
+		t.Fatalf("task repository: %v", err)
+	}
+	t.Cleanup(func() { _ = cleanup() })
+	// agents/agent_profiles live in the settings store; officeRepo's runs
+	// queue (GetRunWorkspaceID) joins against them.
+	if _, _, err := settingsstore.Provide(database, database, nil); err != nil {
+		t.Fatalf("settings store: %v", err)
+	}
+	officeRepo, err := officesqlite.NewWithDB(database, database, nil)
+	if err != nil {
+		t.Fatalf("office repository: %v", err)
+	}
+	log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json", OutputPath: "stdout"})
+	if err != nil {
+		t.Fatalf("logger: %v", err)
+	}
+	taskSvc := taskservice.NewService(taskservice.Repos{
+		Workspaces:       taskRepo,
+		Tasks:            taskRepo,
+		TaskRepos:        taskRepo,
+		Workflows:        taskRepo,
+		Messages:         taskRepo,
+		Turns:            taskRepo,
+		Sessions:         taskRepo,
+		GitSnapshots:     taskRepo,
+		RepoEntities:     taskRepo,
+		Executors:        taskRepo,
+		Environments:     taskRepo,
+		TaskEnvironments: taskRepo,
+		Reviews:          taskRepo,
+		ResourceCleanups: taskRepo,
+	}, bus.NewMemoryEventBus(log), log, taskservice.RepositoryDiscoveryConfig{})
+	return taskSvc, taskRepo, officeRepo
+}
+
+// seedOwnedWorkspace creates a real workspace row owned by ownerID and
+// returns its id. Driven through taskSvc.CreateWorkspace on an identity-free
+// context so the request-body OwnerID is honored (callerScope returns
+// unscoped, exactly the "internal caller" path CreateWorkspace documents) --
+// this seeds a real workspaces row rather than hand-writing SQL, so it stays
+// correct if the schema changes.
+func seedOwnedWorkspace(t *testing.T, taskSvc *taskservice.Service, ownerID string) string {
+	t.Helper()
+	workspace, err := taskSvc.CreateWorkspace(context.Background(), &taskservice.CreateWorkspaceRequest{
+		Name:    "ws-" + ownerID,
+		OwnerID: ownerID,
+	})
+	if err != nil {
+		t.Fatalf("seed workspace owned by %q: %v", ownerID, err)
+	}
+	return workspace.ID
+}
+
+// seedRunForWorkspace inserts an agent + agent profile scoped to
+// workspaceID (empty string reproduces an ordinary, non-Office Kanban agent
+// profile — schema default, never backfilled) and a run against it,
+// returning the run id.
+func seedRunForWorkspace(t *testing.T, officeRepo *officesqlite.Repository, workspaceID string) string {
+	t.Helper()
+	ctx := context.Background()
+	now := time.Now().UTC()
+	agentID := "agent-" + workspaceID
+	if workspaceID == "" {
+		agentID = "agent-unscoped"
+	}
+	if _, err := officeRepo.ExecRaw(ctx, `INSERT INTO agents
+		(id, name, workspace_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?)`,
+		agentID, agentID, workspaceID, now, now); err != nil {
+		t.Fatalf("seed agent: %v", err)
+	}
+	if _, err := officeRepo.ExecRaw(ctx, `INSERT INTO agent_profiles
+		(id, agent_id, name, agent_display_name, workspace_id, role, created_at, updated_at)
+		VALUES (?, ?, ?, ?, ?, '', ?, ?)`,
+		agentID, agentID, "name", "display", workspaceID, now, now); err != nil {
+		t.Fatalf("seed agent profile: %v", err)
+	}
+	runID := "run-" + agentID
+	if err := officeRepo.CreateRun(ctx, &models.Run{
+		ID:             runID,
+		AgentProfileID: agentID,
+		Reason:         "task_assigned",
+		Status:         models.RunStatus("queued"),
+		CoalescedCount: 1,
+	}); err != nil {
+		t.Fatalf("seed run: %v", err)
+	}
+	return runID
+}
+
+// TestRunSubscriptionCheckDeniesEmptyResolvedWorkspace pins WO-02 round 2's
+// FINDING R1-1 fix: an ordinary (non-Office) Kanban agent profile resolves
+// to workspace_id=” via GetRunWorkspaceID, which used to fall into
+// AuthorizeWorkspaceAccess's unconditional-allow branch (workspaceID == ""
+// => nil). The hook must deny outright instead of delegating — this is the
+// dominant class of runs (workflow-engine queue_run against ordinary Kanban
+// agent profiles), not an edge case.
+func TestRunSubscriptionCheckDeniesEmptyResolvedWorkspace(t *testing.T) {
+	taskSvc, _, officeRepo := newRunSubscriptionCheckHarness(t)
+	runID := seedRunForWorkspace(t, officeRepo, "")
+
+	check := runSubscriptionCheck(taskSvc, officeRepo)
+	if err := check(context.Background(), runID); err == nil {
+		t.Fatal("runSubscriptionCheck(empty workspace) = nil, want a denial error")
+	}
+}
+
+// TestRunSubscriptionCheckAllowsResolvedWorkspace drives runSubscriptionCheck
+// with the real WORKSPACE OWNER's scoped identity against a real owned
+// workspace row, so the allow comes from AuthorizeWorkspaceAccess actually
+// reading the workspace and matching OwnerID -- not from callerScope's
+// unscoped short-circuit (authorizeWorkspaceID returns nil before ever
+// reading the row when the context carries no identity, which made the
+// original version of this test pass even with `return nil` substituted for
+// AuthorizeWorkspaceAccess; see REVIEW ROUND 2 FINDING B2-1).
+func TestRunSubscriptionCheckAllowsResolvedWorkspace(t *testing.T) {
+	taskSvc, _, officeRepo := newRunSubscriptionCheckHarness(t)
+	ownerID := "user-owner"
+	workspaceID := seedOwnedWorkspace(t, taskSvc, ownerID)
+	runID := seedRunForWorkspace(t, officeRepo, workspaceID)
+
+	check := runSubscriptionCheck(taskSvc, officeRepo)
+	ctx := authn.WithIdentity(context.Background(), authn.Identity{UserID: ownerID, Role: authn.RoleMember})
+	if err := check(ctx, runID); err != nil {
+		t.Fatalf("runSubscriptionCheck(owner) = %v, want nil", err)
+	}
+}
+
+// TestRunSubscriptionCheckDeniesForeignWorkspaceOwner is the ownership-denial
+// half of B2-1's fix and directly verifies the card's own SCOPE item 4 ("a
+// subscribe for a run the caller does not own must be refused") against the
+// real runSubscriptionCheck -- the only existing foreign-owner coverage
+// (gateway TestSubscriptionChecksDenyForeignTopics) exercises a hand-written
+// fake Run hook, not this one.
+func TestRunSubscriptionCheckDeniesForeignWorkspaceOwner(t *testing.T) {
+	taskSvc, _, officeRepo := newRunSubscriptionCheckHarness(t)
+	workspaceID := seedOwnedWorkspace(t, taskSvc, "user-owner")
+	runID := seedRunForWorkspace(t, officeRepo, workspaceID)
+
+	check := runSubscriptionCheck(taskSvc, officeRepo)
+	ctx := authn.WithIdentity(context.Background(), authn.Identity{UserID: "user-other", Role: authn.RoleMember})
+	if err := check(ctx, runID); err == nil {
+		t.Fatal("runSubscriptionCheck(foreign owner) = nil, want a denial error")
+	}
+}
+
+// TestRunSubscriptionCheckDeniesWhenOfficeRepoNil pins FINDING R1-2: the
+// nil-officeRepo guard must fail closed, not open. Not reachable in the
+// shipped binary (repos.Office is never nil), but the wrong default for a
+// security check.
+func TestRunSubscriptionCheckDeniesWhenOfficeRepoNil(t *testing.T) {
+	taskSvc, _, _ := newRunSubscriptionCheckHarness(t)
+
+	check := runSubscriptionCheck(taskSvc, nil)
+	if err := check(context.Background(), "any-run"); err == nil {
+		t.Fatal("runSubscriptionCheck(nil officeRepo) = nil, want a denial error")
+	}
+}
+
+// TestGatewayAuthPolicyWiresRunSubscriptionCheck pins the production wiring
+// site itself (auth.go's Subscriptions.Run: runSubscriptionCheck(...)),
+// which nothing previously asserted: deleting that one line left every
+// runSubscriptionCheck/gateway test green (B2-1). Scoped to the Run hook
+// only, per the routing message -- Task/Session have no precedent test
+// either and are not part of this card.
+//
+// Beyond the nil check, it also invokes the wired hook against a real
+// empty-workspace run and asserts a denial -- this additionally catches a
+// no-op substitution (e.g. `func(context.Context, string) error { return
+// nil }`) that a bare non-nil assertion would miss.
+func TestGatewayAuthPolicyWiresRunSubscriptionCheck(t *testing.T) {
+	taskSvc, taskRepo, officeRepo := newRunSubscriptionCheckHarness(t)
+	cfg := &config.Config{}
+	cfg.Features.Auth = true
+	cfg.Auth.SessionTTLHours = 720
+	authSvc := newEnabledAuthService(t, cfg)
+
+	policy := gatewayAuthPolicy(authSvc, taskSvc, taskRepo, officeRepo)
+	if policy.Subscriptions.Run == nil {
+		t.Fatal("gatewayAuthPolicy(...).Subscriptions.Run = nil, want runSubscriptionCheck wired")
+	}
+
+	runID := seedRunForWorkspace(t, officeRepo, "")
+	if err := policy.Subscriptions.Run(context.Background(), runID); err == nil {
+		t.Fatal("gatewayAuthPolicy(...).Subscriptions.Run(empty workspace) = nil, want a denial error")
+	}
+}

apps/backend/internal/backendapp/main.go

@@ -815,7 +815,7 @@ func startGatewayAndServe(
 	// and proxy routes, subscription visibility checks, and workspace-owner
 	// broadcast routing. Must be installed before SetupRoutes runs in
 	// buildHTTPServer.
-	gateway.SetAuthPolicy(gatewayAuthPolicy(services.Auth, services.Task, repos.Task))
+	gateway.SetAuthPolicy(gatewayAuthPolicy(services.Auth, services.Task, repos.Task, repos.Office))
 
 	waitForAgentctlControlHealthy(ctx, cfg, log)
 

apps/backend/internal/gateway/websocket/access.go

@@ -20,6 +20,9 @@ import (
 type SubscriptionAccessPolicy struct {
 	Task    func(ctx context.Context, taskID string) error
 	Session func(ctx context.Context, sessionID string) error
+	// Run gates run.subscribe. Nil (the zero-AuthPolicy case) leaves every
+	// run subscription unchecked, matching Task/Session.
+	Run func(ctx context.Context, runID string) error
 }
 
 // WorkspaceOwnerResolver resolves a workspace's owning user ID ("" = unowned

apps/backend/internal/gateway/websocket/access_test.go

@@ -628,6 +628,13 @@ func TestSubscriptionChecksDenyForeignTopics(t *testing.T) {
 				}
 				return nil
 			},
+			Run: func(ctx context.Context, runID string) error {
+				identity, _ := authn.IdentityFromContext(ctx)
+				if runID == "run-b" && identity.UserID != "user-b" {
+					return denied
+				}
+				return nil
+			},
 		},
 	})
 	clientA := registerAccessClient(t, hub, "a", authn.Identity{UserID: "user-a", Role: authn.RoleMember})
@@ -649,12 +656,126 @@ func TestSubscriptionChecksDenyForeignTopics(t *testing.T) {
 		t.Fatal("denied session subscription must not register")
 	}
 
+	clientA.handleRunSubscribe(subscribe(ws.ActionRunSubscribe, map[string]interface{}{"run_id": "run-b"}))
+	assertErrorResponse(t, clientA, "run subscribe")
+	if len(hub.getSubscribersLocked(hub.runSubscribers, "run-b")) != 0 {
+		t.Fatal("denied run subscription must not register")
+	}
+
 	// Own topics still work.
 	clientA.handleSubscribe(subscribe(ws.ActionTaskSubscribe, map[string]interface{}{"task_id": "task-a"}))
 	assertSuccessResponse(t, clientA, "own task subscribe")
 	if len(hub.getSubscribersLocked(hub.taskSubscribers, "task-a")) != 1 {
 		t.Fatal("allowed task subscription must register")
 	}
+
+	clientA.handleRunSubscribe(subscribe(ws.ActionRunSubscribe, map[string]interface{}{"run_id": "run-a"}))
+	assertSuccessResponse(t, clientA, "own run subscribe")
+	if len(hub.getSubscribersLocked(hub.runSubscribers, "run-a")) != 1 {
+		t.Fatal("allowed run subscription must register")
+	}
+}
+
+// TestRunSubscribeDeniesEmptyWorkspaceResolution pins WO-02 round 2's
+// FINDING R1-1 fix at the gateway wiring level: a Run hook denial for a run
+// that resolves to an empty workspace (the runSubscriptionCheck case, an
+// ordinary non-Office agent profile) must refuse the subscribe request and
+// leave it unregistered in hub.runSubscribers — the same contract already
+// pinned for a foreign-owned run by TestSubscriptionChecksDenyForeignTopics,
+// exercised here for the empty-workspace input specifically.
+func TestRunSubscribeDeniesEmptyWorkspaceResolution(t *testing.T) {
+	hub := newAccessTestHub(t)
+	errEmptyWorkspace := errors.New("workspace not found")
+	hub.setAuthPolicy(AuthPolicy{
+		Subscriptions: SubscriptionAccessPolicy{
+			Run: func(_ context.Context, runID string) error {
+				if runID == "run-unscoped" {
+					return errEmptyWorkspace
+				}
+				return nil
+			},
+		},
+	})
+	client := registerAccessClient(t, hub, "a", authn.Identity{UserID: "user-a", Role: authn.RoleMember})
+
+	subscribe := func(runID string) *ws.Message {
+		raw, _ := json.Marshal(map[string]interface{}{"run_id": runID})
+		return &ws.Message{ID: "1", Type: ws.MessageTypeRequest, Action: ws.ActionRunSubscribe, Payload: raw}
+	}
+
+	client.handleRunSubscribe(subscribe("run-unscoped"))
+	assertErrorResponse(t, client, "run subscribe with empty-workspace resolution")
+	if len(hub.getSubscribersLocked(hub.runSubscribers, "run-unscoped")) != 0 {
+		t.Fatal("denied run subscription (empty workspace) must not register")
+	}
+}
+
+// TestRunSubscribeNoExistenceLeak pins the no-existence-leak property: an
+// unknown run id and a foreign run id must be byte-identical on the wire.
+// handleRunSubscribe maps every hook error to one fixed (code, message)
+// pair rather than forwarding err.Error(), so a caller cannot distinguish
+// "you don't own this" from "this doesn't exist" by probing.
+func TestRunSubscribeNoExistenceLeak(t *testing.T) {
+	hub := newAccessTestHub(t)
+	errUnknown := errors.New("sql: no rows in result set")
+	errForeign := errors.New("run belongs to another workspace")
+	hub.setAuthPolicy(AuthPolicy{
+		Subscriptions: SubscriptionAccessPolicy{
+			Run: func(_ context.Context, runID string) error {
+				if runID == "run-unknown" {
+					return errUnknown
+				}
+				return errForeign
+			},
+		},
+	})
+	client := registerAccessClient(t, hub, "a", authn.Identity{UserID: "user-a", Role: authn.RoleMember})
+
+	subscribe := func(id, runID string) *ws.Message {
+		raw, _ := json.Marshal(map[string]interface{}{"run_id": runID})
+		return &ws.Message{ID: id, Type: ws.MessageTypeRequest, Action: ws.ActionRunSubscribe, Payload: raw}
+	}
+
+	client.handleRunSubscribe(subscribe("1", "run-unknown"))
+	unknownFrame := readErrorFrame(t, client, "unknown run")
+
+	client.handleRunSubscribe(subscribe("2", "run-foreign"))
+	foreignFrame := readErrorFrame(t, client, "foreign run")
+
+	if unknownFrame.code != foreignFrame.code || unknownFrame.message != foreignFrame.message {
+		t.Fatalf("unknown run response (%q, %q) must match foreign run response (%q, %q)",
+			unknownFrame.code, unknownFrame.message, foreignFrame.code, foreignFrame.message)
+	}
+}
+
+type errorFrame struct {
+	code    string
+	message string
+}
+
+// readErrorFrame reads one frame off the client's queues and requires it to
+// be an error, decoding the (code, message) payload for comparison.
+func readErrorFrame(t *testing.T, client *Client, label string) errorFrame {
+	t.Helper()
+	var raw []byte
+	select {
+	case raw = <-client.controlSend:
+	case raw = <-client.send:
+	case <-time.After(2 * time.Second):
+		t.Fatalf("%s: no response frame", label)
+	}
+	var msg ws.Message
+	if err := json.Unmarshal(raw, &msg); err != nil {
+		t.Fatalf("%s: bad frame: %v", label, err)
+	}
+	if msg.Type != ws.MessageTypeError {
+		t.Fatalf("%s: got %s frame, want error", label, msg.Type)
+	}
+	var payload ws.ErrorPayload
+	if err := json.Unmarshal(msg.Payload, &payload); err != nil {
+		t.Fatalf("%s: bad error payload: %v", label, err)
+	}
+	return errorFrame{code: payload.Code, message: payload.Message}
 }
 
 func TestUserSubscribeScopedToOwnUser(t *testing.T) {

apps/backend/internal/gateway/websocket/client.go

@@ -502,6 +502,18 @@ func (c *Client) handleRunSubscribe(msg *ws.Message) {
 		c.sendError(msg.ID, msg.Action, ws.ErrorCodeValidation, "run_id is required", nil)
 		return
 	}
+
+	// Per-user scoping: a client may only observe runs in its own
+	// workspaces. Every hook error maps to the same fixed (code, message)
+	// pair below so an unknown run id and a foreign run id are
+	// indistinguishable on the wire (no existence leak).
+	if check := c.hub.authPolicy.Subscriptions.Run; check != nil {
+		if err := check(c.dispatchContext(), req.RunID); err != nil {
+			c.sendError(msg.ID, msg.Action, ws.ErrorCodeForbidden, "cannot subscribe to this run", nil)
+			return
+		}
+	}
+
 	c.hub.SubscribeToRun(c, req.RunID)
 	resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{
 		"success": true,

apps/backend/internal/runs/repository/sqlite/runs.go

@@ -177,6 +177,25 @@ func (r *Repository) GetRunByID(ctx context.Context, id string) (*models.Run, er
 	return &run, nil
 }
 
+// GetRunWorkspaceID resolves a run's owning workspace via its agent
+// profile, the same join idiom ListRuns uses. A raw join, not
+// GetAgentInstance: GetAgentInstance filters deleted_at IS NULL, which
+// would make a run under a soft-deleted agent profile permanently
+// unresolvable (and so permanently deniable) to its own owner. Returns
+// sql.ErrNoRows for an unknown run.
+func (r *Repository) GetRunWorkspaceID(ctx context.Context, runID string) (string, error) {
+	var workspaceID string
+	err := r.ro.QueryRowxContext(ctx, r.ro.Rebind(`
+		SELECT a.workspace_id FROM runs r
+		JOIN agent_profiles a ON a.id = r.agent_profile_id
+		WHERE r.id = ?
+	`), runID).Scan(&workspaceID)
+	if err != nil {
+		return "", err
+	}
+	return workspaceID, nil
+}
+
 // GetClaimedRunByID returns a run only while it is still claimed. Lifecycle
 // events carry this immutable run identity so a delayed predecessor event
 // cannot finish a newer claimed run for the same task and agent.

apps/backend/internal/runs/repository/sqlite/runs_lookup_test.go

@@ -258,3 +258,71 @@ func checkInt64(t *testing.T, field string, got, want int64) {
 		t.Errorf("%s = %d, want %d", field, got, want)
 	}
 }
+
+// TestGetRunWorkspaceID_ResolvesViaAgentProfileJoin pins the WS gateway's
+// run.subscribe authorization lookup (WO-02): the run's owning workspace
+// comes from its agent profile, the same join idiom ListRuns uses.
+func TestGetRunWorkspaceID_ResolvesViaAgentProfileJoin(t *testing.T) {
+	repo, writer := newTestRepoWithDB(t)
+	ctx := context.Background()
+
+	seedAgentProfile(t, writer, "a1", "ws-1")
+	run := seedTaskRun(t, repo, "run-1", "a1", "t1", "queued")
+
+	got, err := repo.GetRunWorkspaceID(ctx, run.ID)
+	if err != nil {
+		t.Fatalf("get run workspace id: %v", err)
+	}
+	checkString(t, "workspace_id", got, "ws-1")
+}
+
+// TestGetRunWorkspaceID_StillResolvesWhenAgentProfileSoftDeleted pins the
+// deliberate choice of a raw join over GetAgentInstance: GetAgentInstance
+// filters deleted_at IS NULL, which would make a run under a soft-deleted
+// agent profile permanently unresolvable (and so permanently deniable) to
+// its own owner.
+func TestGetRunWorkspaceID_StillResolvesWhenAgentProfileSoftDeleted(t *testing.T) {
+	repo, writer := newTestRepoWithDB(t)
+	ctx := context.Background()
+
+	seedAgentProfile(t, writer, "a1", "ws-1")
+	run := seedTaskRun(t, repo, "run-1", "a1", "t1", "queued")
+
+	if _, err := writer.Exec(`UPDATE agent_profiles SET deleted_at = ? WHERE id = 'a1'`, time.Now().UTC()); err != nil {
+		t.Fatalf("soft-delete agent profile: %v", err)
+	}
+
+	got, err := repo.GetRunWorkspaceID(ctx, run.ID)
+	if err != nil {
+		t.Fatalf("get run workspace id after soft delete: %v", err)
+	}
+	checkString(t, "workspace_id", got, "ws-1")
+}
+
+// TestGetRunWorkspaceID_UnknownRunReturnsNoRows pins the sentinel the
+// gateway's subscribe hook denies on.
+func TestGetRunWorkspaceID_UnknownRunReturnsNoRows(t *testing.T) {
+	repo := newTestRepo(t)
+	if _, err := repo.GetRunWorkspaceID(context.Background(), "nope"); !errors.Is(err, sql.ErrNoRows) {
+		t.Fatalf("err = %v, want sql.ErrNoRows", err)
+	}
+}
+
+// TestGetRunWorkspaceID_ResolvesToEmptyStringForUnscopedAgentProfile pins the
+// WO-02 round-2 fix: an ordinary (non-Office) agent profile has
+// workspace_id=” (schema default, no backfill), and the join returns that
+// empty string as-is rather than an error. The gateway's subscribe hook (not
+// this repository) is responsible for denying on an empty result.
+func TestGetRunWorkspaceID_ResolvesToEmptyStringForUnscopedAgentProfile(t *testing.T) {
+	repo, writer := newTestRepoWithDB(t)
+	ctx := context.Background()
+
+	seedAgentProfile(t, writer, "a1", "")
+	run := seedTaskRun(t, repo, "run-1", "a1", "t1", "queued")
+
+	got, err := repo.GetRunWorkspaceID(ctx, run.ID)
+	if err != nil {
+		t.Fatalf("get run workspace id: %v", err)
+	}
+	checkString(t, "workspace_id", got, "")
+}

docs/decisions/2026-07-24-opt-in-authentication.md

@@ -85,7 +85,8 @@ out via `webhooks[].public: true`.
   `auth_identities`; no migration required.
 - Office workspace-scoped HTTP routes are ownership-gated via a group
   middleware (agent-JWT callers keep their workspace-claim scoping); office
-  run subscriptions remain unchecked — accepted gap, noted in the spec.
+  run subscriptions (`run.subscribe`) are now ownership-gated too, closed by
+  WO-02 — see `apps/backend/AGENTS.md` and the spec.
 - The session-access check is enforced at the lifecycle chokepoint AND at
   the reverse-proxy handlers (vscode/port) that resolve executions by bare
   lookup; message read/search and repository-script routes carry explicit

docs/specs/office/live-updates.md

@@ -288,7 +288,7 @@ The kandev backend runs in a single-user model (`store.DefaultUserID = "default-
 
 - **Workspace scoping** — every office WS notification carries `workspace_id` in its payload (when scoped). The current implementation broadcasts office notifications to every connected client and the frontend filters by `workspace_id === workspaces.activeId`. The observable contract is: a client viewing workspace A MUST NOT act on events originating in workspace B. Either server-side filtering or client-side filtering satisfies the contract.
 - **User subscription** — `user.subscribe` accepts only the caller's own user id. Subscribing to another user's id returns WS error `forbidden` (`ws.ErrorCodeForbidden`, "cannot subscribe to another user"). Symmetric rule for `user.unsubscribe`.
-- **Task / session / run subscriptions** — no per-key authorization in the current implementation. Any connected client can subscribe to any `task_id`, `session_id`, or `run_id`. Information-leak risk is bounded by the single-user model. A future multi-user iteration MUST gate `task.subscribe` / `session.subscribe` / `run.subscribe` on workspace membership; the gate point is `Hub.SubscribeToTask` etc.
+- **Task / session / run subscriptions** — auth-enabled connections apply resource-specific access hooks before registration. Task and session use `AuthorizeTaskAccess` and `AuthorizeSessionAccess`. Runs resolve workspace and use `AuthorizeWorkspaceAccess`, denying an empty workspace. Auth-disabled connections retain single-user behavior. See `apps/backend/AGENTS.md`.
 - **Agent-originated subscriptions** — agentctl-initiated MCP tool calls reach the backend over a tunnelled WS connection but do **not** carry a separate identity over this surface; agents cannot subscribe to other agents' streams because subscription frames are scoped per-socket and agents never open the user-facing office WS.
 - **Subscription requests are not approvals** — there is no approval / review gate on subscribing. Once authorized to connect, a client may subscribe and unsubscribe freely.
 

docs/specs/spec-lint.json

@@ -25,7 +25,7 @@
     "docs/specs/office/agents.md": 61654,
     "docs/specs/office/automations-settings.md": 33607,
     "docs/specs/office/costs.md": 33263,
-    "docs/specs/office/live-updates.md": 38602,
+    "docs/specs/office/live-updates.md": 38568,
     "docs/specs/office/overview.md": 34375,
     "docs/specs/office/routing.md": 40782,
     "docs/specs/office/scheduler.md": 50967,