From b37fe6cd5e35e68eb6369d1e8e12b30fca060cf4 Mon Sep 17 00:00:00 2001 From: JJ Eaton <58495321+oldbettie@users.noreply.github.com> Date: Fri, 30 Oct 2026 17:43:12 +0700 Subject: [PATCH] feat(server): tailnet device access syncs across devices Each device keeps a replica of what every tailnet device may do, merged last-writer-wins by hybrid logical clock. Requests are held to the narrower of the token and the current entry, revoking ends a device's sessions and refuses its handshakes everywhere, and any full-access machine can restore it. Writes from revoked or limited devices, or about the writer itself, are ignored. t3_environment_links lists devices, and t3_environment_device_access sets access over the same service method as the RPC. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/auth/EnvironmentAuth.ts | 56 ++- apps/server/src/auth/SessionStore.ts | 39 +++ ...OrchestratorMcpToolkit.integration.test.ts | 2 + .../src/mcp/agents/remoteThreads.test.ts | 2 + apps/server/src/mcp/linkOrigin.test.ts | 2 + apps/server/src/mcp/toolkits/core.test.ts | 6 + .../src/mcp/toolkits/orchestrator/handlers.ts | 21 ++ .../src/mcp/toolkits/orchestrator/tools.ts | 22 +- apps/server/src/peer/DevicePolicy.test.ts | 56 +++ apps/server/src/peer/DevicePolicy.ts | 328 ++++++++++++++++++ apps/server/src/peer/PeerForwarding.test.ts | 77 +++- apps/server/src/peer/PeerForwarding.ts | 19 +- apps/server/src/peer/RemoteDelegation.test.ts | 3 + apps/server/src/peer/TailnetDevices.test.ts | 127 +++++++ apps/server/src/peer/TailnetDevices.ts | 147 ++++---- apps/server/src/persistence/AuthSessions.ts | 36 ++ apps/server/src/persistence/Migrations.ts | 2 + .../fork/013_TailnetDevicePolicy.ts | 16 + .../src/provider/AgentContextBudget.test.ts | 2 + .../src/provider/agentContextBudget.json | 2 +- apps/server/src/ws.ts | 4 +- docs/internals/remote.md | 9 + evals/mcp-tool-use/cases/orchestration.ndjson | 2 +- packages/contracts/src/orchestratorMcp.ts | 34 ++ packages/shared/src/t3McpToolPresentation.ts | 4 + 25 files changed, 932 insertions(+), 86 deletions(-) create mode 100644 apps/server/src/peer/DevicePolicy.test.ts create mode 100644 apps/server/src/peer/DevicePolicy.ts create mode 100644 apps/server/src/persistence/Migrations/fork/013_TailnetDevicePolicy.ts diff --git a/apps/server/src/auth/EnvironmentAuth.ts b/apps/server/src/auth/EnvironmentAuth.ts index 3903da594818a..322a3bc382d33 100644 --- a/apps/server/src/auth/EnvironmentAuth.ts +++ b/apps/server/src/auth/EnvironmentAuth.ts @@ -40,6 +40,7 @@ import * as ServerConfig from "../config.ts"; import * as EnvironmentAuthPolicy from "./EnvironmentAuthPolicy.ts"; import * as PairingGrantStore from "./PairingGrantStore.ts"; import * as ServerSecretStore from "./ServerSecretStore.ts"; +import * as DevicePolicy from "../peer/DevicePolicy.ts"; import * as SessionStore from "./SessionStore.ts"; import { REUSABLE_DEV_SESSION_EXPIRES_AT, resolveReusableDevAuth } from "./ReusableDevAuth.ts"; import { verifyRequestDpopProof } from "./dpop.ts"; @@ -703,6 +704,7 @@ export const make = Effect.gen(function* () { const policy = yield* EnvironmentAuthPolicy.EnvironmentAuthPolicy; const bootstrapCredentials = yield* PairingGrantStore.PairingGrantStore; const sessions = yield* SessionStore.SessionStore; + const devicePolicy = yield* DevicePolicy.DevicePolicy; const secretStore = yield* ServerSecretStore.ServerSecretStore; const crypto = yield* Crypto.Crypto; const descriptor = yield* policy.getDescriptor(); @@ -1283,23 +1285,42 @@ export const make = Effect.gen(function* () { if (token === null) return Effect.fail(new ServerAuthMissingCredentialError({})); return sessions.verify(token).pipe( mapSessionVerificationErrors, - Effect.flatMap((session) => - session.subject === MCP_CLIENT_SUBJECT && session.method === "bearer-access-token" - ? Effect.succeed({ - sessionId: session.sessionId, - label: session.client.label ?? "MCP client", - access: session.scopes.includes(AuthOrchestrationOperateScope) - ? (session.runtimeModeCeiling ?? "approval-required") - : "read-only", - peerLink: session.peerLink === true, - peerEnvironmentId: session.peerEnvironmentId, - } satisfies McpClientSession) - : Effect.fail( - new ServerAuthInvalidCredentialError({ - diagnostic: "Only MCP client sessions are accepted here.", - }), - ), - ), + Effect.flatMap((session) => { + if (session.subject !== MCP_CLIENT_SUBJECT || session.method !== "bearer-access-token") { + return Effect.fail( + new ServerAuthInvalidCredentialError({ + diagnostic: "Only MCP client sessions are accepted here.", + }), + ); + } + const client: McpClientSession = { + sessionId: session.sessionId, + label: session.client.label ?? "MCP client", + access: session.scopes.includes(AuthOrchestrationOperateScope) + ? (session.runtimeModeCeiling ?? "approval-required") + : "read-only", + peerLink: session.peerLink === true, + peerEnvironmentId: session.peerEnvironmentId, + }; + if (session.peerEnvironmentId === undefined) return Effect.succeed(client); + // The device's current policy narrows what its token was issued with, on every request. + return devicePolicy.entryAccess(session.peerEnvironmentId).pipe( + Effect.flatMap((entry) => + Option.isNone(entry) + ? Effect.succeed(client) + : entry.value === "revoked" + ? Effect.fail( + new ServerAuthInvalidCredentialError({ + diagnostic: "This device's access was revoked.", + }), + ) + : Effect.succeed({ + ...client, + access: DevicePolicy.narrowerAccess(client.access, entry.value), + }), + ), + ); + }), Effect.withSpan("EnvironmentAuth.authenticateMcpClient"), ); }; @@ -1395,6 +1416,7 @@ export const make = Effect.gen(function* () { }); export const layer = Layer.effect(EnvironmentAuth, make).pipe( + Layer.provideMerge(DevicePolicy.layer), Layer.provideMerge(PairingGrantStore.layer), Layer.provideMerge(SessionStore.layer), Layer.provideMerge(EnvironmentAuthPolicy.layer), diff --git a/apps/server/src/auth/SessionStore.ts b/apps/server/src/auth/SessionStore.ts index 749c9d26f63d5..78e3c99384dae 100644 --- a/apps/server/src/auth/SessionStore.ts +++ b/apps/server/src/auth/SessionStore.ts @@ -354,6 +354,18 @@ export class OtherSessionsRevocationError extends Schema.TaggedError()( + "PeerSessionsRevocationError", + { + peerEnvironmentId: Schema.String, + ...sessionCredentialInternalErrorContext, + }, +) { + override get message(): string { + return "Failed to revoke another environment's sessions."; + } +} + export const SessionCredentialInternalError = Schema.Union([ SessionClaimsEncodingError, SessionCredentialIssueError, @@ -363,6 +375,7 @@ export const SessionCredentialInternalError = Schema.Union([ ActiveSessionsListError, SessionRevocationError, OtherSessionsRevocationError, + PeerSessionsRevocationError, ]); export type SessionCredentialInternalError = typeof SessionCredentialInternalError.Type; @@ -422,6 +435,10 @@ export class SessionStore extends Context.Service< readonly revokeAllExcept: ( sessionId: AuthSessionId, ) => Effect.Effect; + /** Revokes every session another T3 environment holds here. */ + readonly revokeByPeerEnvironment: ( + peerEnvironmentId: EnvironmentId, + ) => Effect.Effect; readonly markConnected: (sessionId: AuthSessionId) => Effect.Effect; readonly markDisconnected: (sessionId: AuthSessionId) => Effect.Effect; } @@ -1051,6 +1068,27 @@ const make = Effect.gen(function* () { return revokedSessionIds.length; }); + const revokeByPeerEnvironment: SessionStore["Service"]["revokeByPeerEnvironment"] = Effect.fn( + "SessionStore.revokeByPeerEnvironment", + )(function* (peerEnvironmentId) { + const revokedAt = yield* DateTime.now; + const revokedSessionIds = yield* authSessions + .revokeByPeerEnvironment({ peerEnvironmentId, revokedAt }) + .pipe( + Effect.mapError((cause) => new PeerSessionsRevocationError({ peerEnvironmentId, cause })), + ); + yield* Ref.update(connectedSessionsRef, (current) => { + const next = new Map(current); + for (const revokedSessionId of revokedSessionIds) next.delete(revokedSessionId); + return next; + }); + yield* Effect.forEach(revokedSessionIds, emitRemoved, { + concurrency: "unbounded", + discard: true, + }); + return revokedSessionIds.length; + }); + const awaitInvalidation = Effect.fn(function* (sessionId: AuthSessionId) { return yield* Effect.scoped( Effect.gen(function* () { @@ -1094,6 +1132,7 @@ const make = Effect.gen(function* () { }, revoke, revokeAllExcept, + revokeByPeerEnvironment, markConnected, markDisconnected, }); diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index c0f92c104f677..3c859425a7946 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -50,6 +50,7 @@ import * as McpInvocationContext from "./McpInvocationContext.ts"; import * as PeerForwarding from "../peer/PeerForwarding.ts"; import * as PeerLinkRequests from "../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../peer/PeerLinks.ts"; +import * as TailnetDevices from "../peer/TailnetDevices.ts"; const layerOrchestratorToolkit = McpHttpServer.layerOrchestratorToolkit.pipe( Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})), @@ -243,6 +244,7 @@ describe("orchestrator MCP toolkit", () => { Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(TailnetDevices.TailnetDevices)({})), Layer.provide(Layer.mock(ProjectService.ProjectService)({})), Layer.provideMerge( SecretRequests.layer.pipe( diff --git a/apps/server/src/mcp/agents/remoteThreads.test.ts b/apps/server/src/mcp/agents/remoteThreads.test.ts index a1b54a8efd747..e8e01345fa451 100644 --- a/apps/server/src/mcp/agents/remoteThreads.test.ts +++ b/apps/server/src/mcp/agents/remoteThreads.test.ts @@ -25,6 +25,7 @@ import * as DocumentText from "../../provider/DocumentText.ts"; import * as PeerForwarding from "../../peer/PeerForwarding.ts"; import * as PeerLinkRequests from "../../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../../peer/PeerLinks.ts"; +import * as TailnetDevices from "../../peer/TailnetDevices.ts"; import * as ThreadLaunch from "../../orchestration-v2/ThreadLaunchService.ts"; import * as ThreadManagement from "../../orchestration-v2/ThreadManagementService.ts"; import * as ManagedProjectFolders from "../../project/ManagedProjectFolders.ts"; @@ -263,6 +264,7 @@ const makeLayer = (macbook: ReturnType, record: TestRecord) Layer.mock(PeerForwarding.PeerForwarding)({}), Layer.mock(PeerLinkRequests.PeerLinkRequests)({}), Layer.mock(PeerLinks.PeerLinks)({ get: () => Effect.succeedNone }), + Layer.mock(TailnetDevices.TailnetDevices)({}), Layer.mock(SourceControlRepositoryService.SourceControlRepositoryService)({}), NodeCrypto.layer, NodeServices.layer, diff --git a/apps/server/src/mcp/linkOrigin.test.ts b/apps/server/src/mcp/linkOrigin.test.ts index c17f4ba579dd6..b1387b3de7197 100644 --- a/apps/server/src/mcp/linkOrigin.test.ts +++ b/apps/server/src/mcp/linkOrigin.test.ts @@ -46,6 +46,7 @@ import * as PeerForwarding from "../peer/PeerForwarding.ts"; import * as ThreadSearch from "../orchestration-v2/ThreadSearch.ts"; import * as PeerLinkRequests from "../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../peer/PeerLinks.ts"; +import * as TailnetDevices from "../peer/TailnetDevices.ts"; // A linked environment's session drives this one's real orchestrator through // its real T3 tools. What the link starts carries its origin, and the link @@ -161,6 +162,7 @@ const layerTools = Layer.mergeAll( Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(TailnetDevices.TailnetDevices)({})), Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})), Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), Layer.provide( diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 4086ed785e045..e741968507612 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -58,6 +58,7 @@ import * as AttachmentHandlers from "./attachment/handlers.ts"; import * as PeerForwarding from "../../peer/PeerForwarding.ts"; import * as PeerLinkRequests from "../../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../../peer/PeerLinks.ts"; +import * as TailnetDevices from "../../peer/TailnetDevices.ts"; const layerOrchestratorToolkit = McpHttpServer.layerOrchestratorToolkit.pipe( Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})), @@ -79,6 +80,7 @@ const layerThreadToolkit = McpHttpServer.layerThreadToolkit.pipe( Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(TailnetDevices.TailnetDevices)({})), ); const threadId = ThreadId.make("mcp-core-thread"); @@ -608,6 +610,7 @@ it.effect("refuses act-as-caller tools to a client caller", () => Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(TailnetDevices.TailnetDevices)({})), Layer.provide(Layer.mock(ProjectService.ProjectService)({})), Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), @@ -669,6 +672,7 @@ it.effect("a caller cannot rewrite a scheduled task that runs above its own mode Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(TailnetDevices.TailnetDevices)({})), Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), ), ), @@ -758,6 +762,7 @@ it.effect.each(["t3_thread_interrupt", "t3_workflow_stop"])( Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(TailnetDevices.TailnetDevices)({})), ), ), ), @@ -851,6 +856,7 @@ it.effect("discovers workflow stop targets and requires orchestration capability Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(TailnetDevices.TailnetDevices)({})), Layer.provide(Layer.mock(ProjectService.ProjectService)({})), Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), diff --git a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts index b70d0d0cd051e..18ac06b203ced 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts @@ -6,6 +6,7 @@ import * as Effect from "effect/Effect"; import * as PeerForwarding from "../../../peer/PeerForwarding.ts"; import * as PeerLinkRequests from "../../../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../../../peer/PeerLinks.ts"; +import * as TailnetDevices from "../../../peer/TailnetDevices.ts"; import { readCaller } from "../../threadAccess.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; import * as McpToolAccess from "../../McpToolAccess.ts"; @@ -236,6 +237,26 @@ export const layer = McpToolAccess.toLayer( return { environmentId, removed }; }), ), + // A device's access holds on every one of the user's machines, so it is + // an environment setting too. + t3_environment_device_access: McpToolAccess.writesEnvironment(({ environmentId, access }) => + Effect.gen(function* () { + const devices = yield* TailnetDevices.TailnetDevices; + const entry = yield* devices.setAccess(environmentId, access).pipe( + Effect.mapError( + (error) => + new OrchestratorMcpFailure({ + code: + error.reason === "unknown_device" || error.reason === "self" + ? "invalid_request" + : "orchestration_error", + message: error.message, + }), + ), + ); + return { environmentId: entry.environmentId, access: entry.access }; + }), + ), t3_thread_interrupt: McpToolAccess.writesThreads( (input) => [input.threadId], (input) => diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index c2740a9346f3f..67a6d6e1383ee 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -2,6 +2,8 @@ import { OrchestratorMcpCapabilitiesInput, OrchestratorMcpCapabilitiesResult, OrchestrationV2StopWorkflowInput, + OrchestratorMcpEnvironmentDeviceAccessInput, + OrchestratorMcpEnvironmentDeviceAccessResult, OrchestratorMcpEnvironmentLinkInput, OrchestratorMcpEnvironmentLinkResult, OrchestratorMcpEnvironmentLinksResult, @@ -47,6 +49,7 @@ import * as ThreadManagementService from "../../../orchestration-v2/ThreadManage import * as PeerForwarding from "../../../peer/PeerForwarding.ts"; import * as PeerLinkRequests from "../../../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../../../peer/PeerLinks.ts"; +import * as TailnetDevices from "../../../peer/TailnetDevices.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; import * as OrchestratorMcpService from "../../OrchestratorMcpService.ts"; import * as ThreadMetadataMcpService from "../../ThreadMetadataMcpService.ts"; @@ -66,6 +69,8 @@ const linkingDependencies = [ PeerLinkRequests.PeerLinkRequests, PeerLinks.PeerLinks, ]; +/** Setting what one of the user's tailnet devices may do. */ +const deviceDependencies = [...dependencies, TailnetDevices.TailnetDevices]; const threadMetadataDependencies = [ McpInvocationContext.McpInvocationContext, ThreadManagementService.ThreadManagementService, @@ -316,7 +321,7 @@ const ThreadInterruptTool = Tool.make("t3_thread_interrupt", { const EnvironmentLinksTool = Tool.make("t3_environment_links", { description: - "List linked environments (the user's other machines) and whether each answers now. Pass one's `environmentId` to a tool to act there; `delegate_task` takes `target.environmentId`. Expired: link it again with `t3_environment_link`.", + "List linked environments (the user's other machines) and whether each answers now. Pass one's `environmentId` to a tool to act there; `delegate_task` takes `target.environmentId`. Expired: link it again with `t3_environment_link`. `devices` lists the user's own machines on their tailnet, linked or not, with what each may do (`revoked`: cut off everywhere).", success: OrchestratorMcpEnvironmentLinksResult, failure: OrchestratorMcpFailure, failureMode: "return", @@ -356,6 +361,20 @@ const EnvironmentUnlinkTool = Tool.make("t3_environment_unlink", { .annotate(Tool.Idempotent, true) .annotate(McpSchema.EnabledWhen, McpInvocationContext.hiddenWhileLinksOff); +const EnvironmentDeviceAccessTool = Tool.make("t3_environment_device_access", { + description: + "Set what one of the user's own machines may do on all of their machines. `revoked` cuts it off everywhere: ends its sessions and refuses its handshakes. Any other level restores it.", + parameters: OrchestratorMcpEnvironmentDeviceAccessInput, + success: OrchestratorMcpEnvironmentDeviceAccessResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies: deviceDependencies, +}) + .annotate(Tool.Title, "Set a device's access") + .annotate(Tool.Destructive, true) + .annotate(Tool.Idempotent, true) + .annotate(McpSchema.EnabledWhen, McpInvocationContext.hiddenWhileLinksOff); + export const OrchestratorToolkit = Toolkit.make( OrchestratorCapabilitiesTool, DelegateTaskTool, @@ -377,4 +396,5 @@ export const OrchestratorToolkit = Toolkit.make( EnvironmentLinksTool, EnvironmentLinkTool, EnvironmentUnlinkTool, + EnvironmentDeviceAccessTool, ); diff --git a/apps/server/src/peer/DevicePolicy.test.ts b/apps/server/src/peer/DevicePolicy.test.ts new file mode 100644 index 0000000000000..05c1a0f405b81 --- /dev/null +++ b/apps/server/src/peer/DevicePolicy.test.ts @@ -0,0 +1,56 @@ +import { EnvironmentId, type DevicePolicyEntry } from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { mergePolicy } from "./DevicePolicy.ts"; + +const laptop = EnvironmentId.make("env-laptop"); +const studio = EnvironmentId.make("env-studio"); +const box = EnvironmentId.make("env-box"); + +const write = ( + environmentId: EnvironmentId, + access: DevicePolicyEntry["access"], + wallMs: number, + counter: number, + by: EnvironmentId, +): DevicePolicyEntry => ({ + environmentId, + access, + updatedAt: { wallMs, counter, node: by }, + updatedBy: by, +}); + +const writes = [ + write(box, "revoked", 10, 0, laptop), + write(box, "full-access", 10, 1, studio), + write(box, "read-only", 10, 1, laptop), + write(studio, "auto", 5, 0, laptop), + write(studio, "revoked", 4, 9, box), +]; + +const permutations = (items: ReadonlyArray): ReadonlyArray> => + items.length <= 1 + ? [items] + : items.flatMap((item, index) => + permutations(items.filter((_, other) => other !== index)).map((rest) => [item, ...rest]), + ); + +describe("mergePolicy", () => { + it("converges on the same policy whatever order writes arrive in", () => { + const results = permutations(writes).map((order) => + // One write at a time, as gossip delivers them. + [ + ...order.reduce>( + (policy, entry) => mergePolicy(policy, [entry]), + new Map(), + ), + ].toSorted(([a], [b]) => a.localeCompare(b)), + ); + + for (const result of results) expect(result).toEqual(results[0]); + expect(Object.fromEntries(results[0]!.map(([id, entry]) => [id, entry.access]))).toEqual({ + [box]: "full-access", + [studio]: "auto", + }); + }); +}); diff --git a/apps/server/src/peer/DevicePolicy.ts b/apps/server/src/peer/DevicePolicy.ts new file mode 100644 index 0000000000000..5dd9ad301a905 --- /dev/null +++ b/apps/server/src/peer/DevicePolicy.ts @@ -0,0 +1,328 @@ +import { + type AuthMcpClientAccess, + DEFAULT_TAILNET_DEVICES_SETTINGS, + DevicePolicyEntry, + type EnvironmentId, + TailnetDeviceAccess, + TailnetDevicesError, + TailnetDevicesSettings, + type TailnetDevicesUpdateSettingsInput, + TailnetHlc, +} from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; +import * as SqlClient from "effect/sql/SqlClient"; + +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; + +const compareStamps = (a: TailnetHlc, b: TailnetHlc): number => + a.wallMs !== b.wallMs + ? a.wallMs - b.wallMs + : a.counter !== b.counter + ? a.counter - b.counter + : a.node < b.node + ? -1 + : a.node > b.node + ? 1 + : 0; + +const ACCESS_ORDER: ReadonlyArray = [ + "read-only", + "approval-required", + "auto-accept-edits", + "auto", + "full-access", +]; + +/** The narrower of two access levels. */ +export const narrowerAccess = (a: AuthMcpClientAccess, b: AuthMcpClientAccess) => + ACCESS_ORDER.indexOf(a) <= ACCESS_ORDER.indexOf(b) ? a : b; + +/** + * The policy after taking `incoming`: each device keeps its latest write, so + * any order of the same writes ends in the same policy. + */ +export const mergePolicy = ( + current: ReadonlyMap, + incoming: ReadonlyArray, +): ReadonlyMap => { + const next = new Map(current); + for (const entry of incoming) { + const existing = next.get(entry.environmentId); + if (existing === undefined || compareStamps(entry.updatedAt, existing.updatedAt) > 0) { + next.set(entry.environmentId, entry); + } + } + return next; +}; + +interface PolicyRow { + readonly environment_id: string; + readonly access: string; + readonly stamp_json: string; + readonly updated_by: string; +} + +const decodeEntry = Schema.decodeUnknownEffect(DevicePolicyEntry); +const isAccess = Schema.is(TailnetDeviceAccess); +const decodeStamp = Schema.decodeUnknownEffect(Schema.fromJsonString(TailnetHlc)); +const encodeStamp = Schema.encodeEffect(Schema.fromJsonString(TailnetHlc)); +const decodeSettings = Schema.decodeUnknownEffect(Schema.fromJsonString(TailnetDevicesSettings)); +const encodeSettings = Schema.encodeEffect(Schema.fromJsonString(TailnetDevicesSettings)); + +const storageError = (cause: unknown) => + new TailnetDevicesError({ + reason: "storage", + message: "Device access could not be read or saved here. The server log has details.", + cause, + }); + +/** + * What each of my tailnet devices may do, and the settings they share. Every + * device holds a replica; a write here is stamped with a hybrid logical clock + * and replicas merge last-writer-wins. Only full-access devices' writes count. + */ +export class DevicePolicy extends Context.Service< + DevicePolicy, + { + readonly settings: Effect.Effect; + readonly updateSettings: ( + input: TailnetDevicesUpdateSettingsInput, + ) => Effect.Effect; + readonly entries: Effect.Effect, TailnetDevicesError>; + /** The access someone set for the device, if anyone did. */ + readonly entryAccess: ( + environmentId: EnvironmentId, + ) => Effect.Effect>; + /** The device's own entry, or the shared default access. */ + readonly accessFor: (environmentId: EnvironmentId) => Effect.Effect; + readonly set: ( + environmentId: EnvironmentId, + access: TailnetDeviceAccess, + ) => Effect.Effect; + /** + * Takes another device's replica, unless that device may not write policy: + * revoked or below full access here. Returns the devices it newly revoked. + */ + readonly merge: ( + sender: EnvironmentId, + replica: { + readonly policy: ReadonlyArray; + readonly settings: TailnetDevicesSettings; + }, + ) => Effect.Effect, TailnetDevicesError>; + } +>()("t3/peer/DevicePolicy") {} + +const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const identity = yield* ServerEnvironment.ServerEnvironmentIdentity; + const lastStamp = yield* Ref.make(null); + // Reads, merges and writes of the replica happen one at a time. + const writes = yield* Semaphore.make(1); + + const settings: DevicePolicy["Service"]["settings"] = sql<{ settings_json: string }>` + SELECT settings_json FROM tailnet_device_settings WHERE id = 1 + `.pipe( + Effect.flatMap((rows) => + rows[0] === undefined + ? Effect.succeed(DEFAULT_TAILNET_DEVICES_SETTINGS) + : decodeSettings(rows[0].settings_json), + ), + Effect.tapError((cause) => Effect.logWarning("Tailnet device settings unreadable", { cause })), + Effect.orElseSucceed(() => DEFAULT_TAILNET_DEVICES_SETTINGS), + ); + + const readEntries = sql`SELECT * FROM tailnet_device_policy`.pipe( + Effect.flatMap((rows) => + Effect.forEach(rows, (row) => + decodeStamp(row.stamp_json).pipe( + Effect.flatMap((updatedAt) => + decodeEntry({ + environmentId: row.environment_id, + access: row.access, + updatedAt, + updatedBy: row.updated_by, + }), + ), + ), + ), + ), + Effect.mapError(storageError), + ); + + /** Later than every stamp this environment made or has seen, so its next write wins. */ + const observe = (stamps: ReadonlyArray) => + Ref.update(lastStamp, (last) => + stamps.reduce( + (latest, stamp) => + stamp !== null && (latest === null || compareStamps(stamp, latest) > 0) ? stamp : latest, + last, + ), + ); + + const stamp = Effect.gen(function* () { + const node = yield* identity.getEnvironmentId; + const now = yield* Clock.currentTimeMillis; + return yield* Ref.modify(lastStamp, (last): [TailnetHlc, TailnetHlc] => { + const next = + last === null || now > last.wallMs + ? { wallMs: now, counter: 0, node } + : { wallMs: last.wallMs, counter: last.counter + 1, node }; + return [next, next]; + }); + }); + + // A restart must not stamp earlier than what is already stored. + yield* Effect.all([readEntries, settings]).pipe( + Effect.flatMap(([stored, current]) => + observe([...stored.map((entry) => entry.updatedAt), current.updatedAt]), + ), + Effect.ignore, + ); + + const writeEntry = (entry: DevicePolicyEntry) => + encodeStamp(entry.updatedAt).pipe( + Effect.flatMap( + (stampJson) => sql` + INSERT INTO tailnet_device_policy (environment_id, access, stamp_json, updated_by) + VALUES (${entry.environmentId}, ${entry.access}, ${stampJson}, ${entry.updatedBy}) + ON CONFLICT (environment_id) DO UPDATE SET + access = excluded.access, + stamp_json = excluded.stamp_json, + updated_by = excluded.updated_by + `, + ), + Effect.mapError(storageError), + ); + + const writeSettings = (next: TailnetDevicesSettings) => + encodeSettings(next).pipe( + Effect.flatMap( + (json) => sql` + INSERT INTO tailnet_device_settings (id, settings_json) VALUES (1, ${json}) + ON CONFLICT (id) DO UPDATE SET settings_json = excluded.settings_json + `, + ), + Effect.mapError(storageError), + ); + + const entryAccess: DevicePolicy["Service"]["entryAccess"] = Effect.fn("DevicePolicy.entryAccess")( + function* (environmentId) { + const rows = yield* sql<{ access: string }>` + SELECT access FROM tailnet_device_policy WHERE environment_id = ${environmentId} + `.pipe(Effect.orElseSucceed(() => [])); + const stored = rows[0]?.access; + if (stored === undefined) return Option.none(); + // An unreadable row fails closed. + return Option.some(isAccess(stored) ? stored : "revoked"); + }, + ); + + const accessFor: DevicePolicy["Service"]["accessFor"] = Effect.fn("DevicePolicy.accessFor")( + function* (environmentId) { + const entry = yield* entryAccess(environmentId); + return Option.isSome(entry) ? entry.value : (yield* settings).defaultAccess; + }, + ); + + const set: DevicePolicy["Service"]["set"] = Effect.fn("DevicePolicy.set")( + function* (environmentId, access) { + const updatedAt = yield* stamp; + const entry: DevicePolicyEntry = { + environmentId, + access, + updatedAt, + updatedBy: updatedAt.node, + }; + yield* writeEntry(entry); + return entry; + }, + (effect) => writes.withPermits(1)(effect), + ); + + const updateSettings: DevicePolicy["Service"]["updateSettings"] = Effect.fn( + "DevicePolicy.updateSettings", + )( + function* (input) { + const current = yield* settings; + const updatedAt = yield* stamp; + const next: TailnetDevicesSettings = { + autoLink: input.autoLink ?? current.autoLink, + defaultAccess: input.defaultAccess ?? current.defaultAccess, + requireCapability: input.requireCapability ?? current.requireCapability, + updatedAt, + updatedBy: updatedAt.node, + }; + yield* writeSettings(next); + return next; + }, + (effect) => writes.withPermits(1)(effect), + ); + + const merge: DevicePolicy["Service"]["merge"] = Effect.fn("DevicePolicy.merge")( + function* (sender, replica) { + // A revoked device cannot restore itself, and a limited one cannot widen anyone. + if ((yield* accessFor(sender)) !== "full-access") return []; + const self = yield* identity.getEnvironmentId; + // The same holds for each write the sender relays: no device sets its own access, + // and only a writer full-access here counts. + const accepted: Array = []; + for (const entry of replica.policy) { + if (entry.environmentId === entry.updatedBy) continue; + if (entry.updatedBy !== self && (yield* accessFor(entry.updatedBy)) !== "full-access") { + continue; + } + accepted.push(entry); + } + const stored = yield* readEntries; + const current = new Map(stored.map((entry) => [entry.environmentId, entry])); + const next = mergePolicy(current, accepted); + const newlyRevoked: Array = []; + for (const [environmentId, entry] of next) { + if (current.get(environmentId) === entry) continue; + yield* writeEntry(entry); + if (entry.access === "revoked" && current.get(environmentId)?.access !== "revoked") { + newlyRevoked.push(environmentId); + } + } + const ownSettings = yield* settings; + const incoming = replica.settings; + const writerCounts = + incoming.updatedBy !== null && + (incoming.updatedBy === self || (yield* accessFor(incoming.updatedBy)) === "full-access"); + if ( + writerCounts && + incoming.updatedAt !== null && + (ownSettings.updatedAt === null || + compareStamps(incoming.updatedAt, ownSettings.updatedAt) > 0) + ) { + yield* writeSettings(incoming); + } + yield* observe([ + ...accepted.map((entry) => entry.updatedAt), + writerCounts ? incoming.updatedAt : null, + ]); + return newlyRevoked; + }, + (effect) => writes.withPermits(1)(effect), + ); + + return DevicePolicy.of({ + settings, + updateSettings, + entries: readEntries, + entryAccess, + accessFor, + set, + merge, + }); +}); + +export const layer = Layer.effect(DevicePolicy, make); diff --git a/apps/server/src/peer/PeerForwarding.test.ts b/apps/server/src/peer/PeerForwarding.test.ts index e16a464f7a8b2..258cd8d76a3b0 100644 --- a/apps/server/src/peer/PeerForwarding.test.ts +++ b/apps/server/src/peer/PeerForwarding.test.ts @@ -3,6 +3,7 @@ import { NodeHttpServer } from "@effect/platform-node"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { + DEFAULT_TAILNET_DEVICES_SETTINGS, EnvironmentId, type OrchestrationV2Run, type OrchestrationV2ThreadShell, @@ -10,6 +11,7 @@ import { ProjectId, ProviderInstanceId, RunId, + type TailnetDeviceAccess, ThreadId, } from "@t3tools/contracts"; import { expect, it } from "@effect/vitest"; @@ -39,10 +41,13 @@ import { liveThreadShell } from "../mcp/McpToolAccess.testkit.ts"; import * as PeerForwarding from "./PeerForwarding.ts"; import * as PeerLinks from "./PeerLinks.ts"; import * as PeerLinkRequests from "./PeerLinkRequests.ts"; +import * as TailnetDevices from "./TailnetDevices.ts"; import { descriptorOf, layerLinkingEnvironment, linkTo, servePeer } from "./PeerLinks.testkit.ts"; const laptop = descriptorOf("environment-laptop", "Laptop"); const box = descriptorOf("environment-box", "Box"); +/** A machine of the user's that is revoked everywhere. */ +const desk = EnvironmentId.make("environment-desk"); const boxProject = ProjectId.make("project:box"); const boxThread = ThreadId.make("thread:box-auto"); const boxRun = RunId.make("run:box-auto"); @@ -233,6 +238,7 @@ const serveBox = (seen: Ref.Ref, linkSession: Ref.Ref) => Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(TailnetDevices.TailnetDevices)({})), ), ); @@ -255,6 +261,38 @@ const mcpClient = McpSchema.McpServerClient.of({ */ const makeLaptop = Effect.gen(function* () { const linking = yield* layerLinkingEnvironment(laptop).pipe(Layer.build); + const accessSet = yield* Ref.make>( + [], + ); + const layerDevices = Layer.mock(TailnetDevices.TailnetDevices)({ + list: Effect.succeed({ + self: { environmentId: laptop.environmentId, label: laptop.label, urls: [] }, + settings: DEFAULT_TAILNET_DEVICES_SETTINGS, + devices: [ + { + environmentId: desk, + label: "Desk", + urls: [], + status: "revoked", + access: "revoked", + policy: null, + outgoing: null, + incoming: false, + tailnetName: "desk", + lastSeenAt: null, + }, + ], + }), + setAccess: (environmentId, access) => + Ref.update(accessSet, (seen) => [...seen, [environmentId, access] as const]).pipe( + Effect.as({ + environmentId, + access, + updatedAt: { wallMs: 1, counter: 0, node: laptop.environmentId }, + updatedBy: laptop.environmentId, + }), + ), + }); const layerHere = Layer.mergeAll( McpHttpServer.layerOrchestratorToolkit, McpHttpServer.layerProjectRegistration, @@ -276,6 +314,7 @@ const makeLaptop = Effect.gen(function* () { ), Layer.provide(NodeServices.layer), Layer.provide(PeerForwarding.layer), + Layer.provide(layerDevices), Layer.provide(Layer.succeedContext(linking)), Layer.provide(NodeCrypto.layer), Layer.provideMerge( @@ -347,7 +386,7 @@ const makeLaptop = Effect.gen(function* () { ), Effect.provideService(McpSchema.McpServerClient, mcpClient), ); - return { links: Context.get(linking, PeerLinks.PeerLinks), call }; + return { links: Context.get(linking, PeerLinks.PeerLinks), call, accessSet }; }); const failureOf = (result: McpSchema.CallToolResult) => { @@ -369,9 +408,33 @@ it.effect("an agent here works on a linked environment's threads within its own expect(listed.structuredContent).toMatchObject({ environmentId: laptop.environmentId, links: [ - { environmentId: box.environmentId, label: "Box", status: "reachable", access: "auto" }, + { + environmentId: box.environmentId, + label: "Box", + status: "reachable", + access: "auto", + origin: "paired", + }, ], + devices: [ + { + environmentId: desk, + label: "Desk", + status: "revoked", + access: "revoked", + outgoing: null, + incoming: false, + }, + ], + }); + + // A full-access agent here may restore one of the user's machines everywhere. + const restored = yield* a.call("thread:laptop-full", "t3_environment_device_access", { + environmentId: desk, + access: "auto", }); + expect(restored.structuredContent).toEqual({ environmentId: desk, access: "auto" }); + expect(yield* Ref.get(a.accessSet)).toEqual([[desk, "auto"]]); const threads = yield* a.call("thread:laptop-full", "t3_thread_list", { environmentId: box.environmentId, @@ -684,6 +747,16 @@ it.effect("refuses what a linked environment cannot take before anything leaves" expect(failureOf(relayed), name).toMatchObject({ code: "capability_denied" }); } expect(yield* Ref.get(b.bearers)).toEqual([]); + + // A device's access holds on every machine: full access only, and never linked work. + for (const caller of ["thread:laptop-plan", "thread:laptop-from-desk"]) { + const refused = yield* a.call(caller, "t3_environment_device_access", { + environmentId: desk, + access: "revoked", + }); + expect(failureOf(refused), caller).toMatchObject({ code: "capability_denied" }); + } + expect(yield* Ref.get(a.accessSet)).toEqual([]); }), ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); diff --git a/apps/server/src/peer/PeerForwarding.ts b/apps/server/src/peer/PeerForwarding.ts index b163f376319bc..075b6ea627a5b 100644 --- a/apps/server/src/peer/PeerForwarding.ts +++ b/apps/server/src/peer/PeerForwarding.ts @@ -24,6 +24,7 @@ import { readCaller, unavailable } from "../mcp/threadAccess.ts"; import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; import * as PeerLinks from "./PeerLinks.ts"; import * as PeerMcpClient from "./PeerMcpClient.ts"; +import * as TailnetDevices from "./TailnetDevices.ts"; /** * A peer's `/mcp` holds a wait open for its whole timeout without sending @@ -94,6 +95,7 @@ export const remoteTarget = (scope: Scope, environmentId: EnvironmentId | undefi const make = Effect.gen(function* () { const peers = yield* PeerMcpClient.PeerMcpClient; const links = yield* PeerLinks.PeerLinks; + const tailnetDevices = yield* TailnetDevices.TailnetDevices; const crypto = yield* Crypto.Crypto; const threads = yield* ThreadManagement.ThreadManagementService; @@ -239,7 +241,7 @@ const make = Effect.gen(function* () { * empty list reads as a dead end and agents go looking elsewhere. */ const NOT_LISTED = - "Not listed: not linked yet. Call t3_environment_link with a hint from the user's words."; + "Not in links: not linked yet. Call t3_environment_link with its environmentId from devices, or a hint from the user's words."; const listLinks: PeerForwarding["Service"]["links"] = Effect.fn("PeerForwarding.listLinks")( function* (scope) { @@ -251,6 +253,12 @@ const make = Effect.gen(function* () { new OrchestratorMcpFailure({ code: "orchestration_error", message: error.message }), ), ); + // Devices only add to the links, so an unreadable tailnet still lists them. + const devices = yield* tailnetDevices.list.pipe( + Effect.map((snapshot) => snapshot.devices), + Effect.tapError((error) => Effect.logWarning("Tailnet devices unreadable", error)), + Effect.orElseSucceed(() => []), + ); return { environmentId: scope.environmentId, links: listed.map((link) => ({ @@ -258,9 +266,18 @@ const make = Effect.gen(function* () { label: link.label, status: link.status, access: link.access, + origin: link.origin, expiresAt: DateTime.formatIso(link.expiresAt), lastError: link.lastError, })), + devices: devices.map((device) => ({ + environmentId: device.environmentId, + label: device.label, + status: device.status, + access: device.access, + outgoing: device.outgoing, + incoming: device.incoming, + })), notListed: NOT_LISTED, }; }, diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index bc56d90561fa5..e2fdf6dacc2cf 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -60,6 +60,7 @@ import * as SourceControlRepositoryService from "../sourceControl/SourceControlR import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as PeerForwarding from "./PeerForwarding.ts"; import * as PeerLinks from "./PeerLinks.ts"; +import * as TailnetDevices from "./TailnetDevices.ts"; import { descriptorOf, layerLinkingEnvironment, linkTo, servePeer } from "./PeerLinks.testkit.ts"; import * as RemoteDelegation from "./RemoteDelegation.ts"; import * as PeerLinkRequests from "./PeerLinkRequests.ts"; @@ -471,6 +472,7 @@ const boxToolkitLayer = (thread: BoxThread) => { Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(TailnetDevices.TailnetDevices)({})), Layer.provide( Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/p" }), ), @@ -587,6 +589,7 @@ const makeLaptop = ( ), ), Layer.provide(Layer.succeedContext(linking)), + Layer.provide(Layer.mock(TailnetDevices.TailnetDevices)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(NodeCrypto.layer), Layer.provideMerge(layerThreads), diff --git a/apps/server/src/peer/TailnetDevices.test.ts b/apps/server/src/peer/TailnetDevices.test.ts index 8128954965053..bd017b82cc9e2 100644 --- a/apps/server/src/peer/TailnetDevices.test.ts +++ b/apps/server/src/peer/TailnetDevices.test.ts @@ -30,6 +30,7 @@ import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as ServerHttp from "../http.ts"; import * as Sqlite from "../persistence/Sqlite.ts"; import * as ServerSettings from "../serverSettings.ts"; +import * as DevicePolicy from "./DevicePolicy.ts"; import * as PeerLinks from "./PeerLinks.ts"; import { descriptorOf } from "./PeerLinks.testkit.ts"; import * as TailnetDevices from "./TailnetDevices.ts"; @@ -180,6 +181,14 @@ const serveDeviceOn = ( descriptor, devices: Context.get(context, TailnetDevices.TailnetDevices), links: Context.get(context, PeerLinks.PeerLinks), + policy: Context.get(context, DevicePolicy.DevicePolicy), + /** What this device lets a caller holding `token` do, as its /mcp would see it. */ + authenticate: (token: string) => + auth.authenticateMcpClient( + HttpServerRequest.fromWeb( + new Request(`${origin}/mcp`, { headers: { authorization: `Bearer ${token}` } }), + ), + ), settings: Context.get(context, ServerSettings.ServerSettingsService), /** Grants delivered to this device, each carrying a token. */ grants, @@ -382,6 +391,124 @@ describe("tailnet device handshake", () => { ); }); +/** Laptop, studio and box, every pair linked both ways. */ +const threeDevices = Effect.gen(function* () { + const tailnet = yield* makeTailnet; + const a = yield* serveDevice(laptop, tailnet); + const b = yield* serveDevice(studio, tailnet); + const c = yield* serveDevice(box, tailnet); + yield* tailnet.join(a.url); + yield* tailnet.join(b.url); + yield* tailnet.join(c.url); + yield* a.devices.linkIfMine({ url: b.url }); + yield* a.devices.linkIfMine({ url: c.url }); + yield* b.devices.linkIfMine({ url: c.url }); + return { a, b, c }; +}); + +const box = descriptorOf("env-box", "Box"); + +describe("tailnet device access", () => { + it.live("applies a changed access level to the next request without re-linking", () => + Effect.gen(function* () { + const tailnet = yield* makeTailnet; + const a = yield* serveDevice(laptop, tailnet); + const b = yield* serveDevice(studio, tailnet); + yield* tailnet.join(a.url); + yield* tailnet.join(b.url); + yield* a.devices.linkIfMine({ url: b.url }); + const { token } = yield* b.links.resolve(laptop.environmentId); + expect((yield* a.authenticate(token)).access).toBe("full-access"); + + yield* a.devices.setAccess(studio.environmentId, "read-only"); + + expect((yield* a.authenticate(token)).access).toBe("read-only"); + }), + ); + + it.live("revoking a device on one machine cuts it off on every machine", () => + Effect.gen(function* () { + const { a, b, c } = yield* threeDevices; + const { token } = yield* c.links.resolve(studio.environmentId); + + yield* a.devices.setAccess(box.environmentId, "revoked"); + + for (const device of [a, b]) { + expect(yield* device.policy.accessFor(box.environmentId)).toBe("revoked"); + expect( + (yield* device.issued).filter((session) => session.peerEnvironmentId === "env-box"), + ).toEqual([]); + expect(yield* device.links.get(box.environmentId)).toEqual(Option.none()); + expect( + (yield* device.devices.list).devices.find((row) => row.environmentId === "env-box") + ?.status, + ).toBe("revoked"); + } + expect(Option.isNone(yield* Effect.option(b.authenticate(token)))).toBe(true); + const grants = yield* Ref.get(c.grants); + expect(yield* requestAs(b, { environmentId: "env-box", urls: [c.url] })).toEqual({ + outcome: "refused", + reason: "revoked", + }); + expect(yield* c.devices.linkIfMine({ url: b.url })).toEqual(Option.none()); + expect(yield* Ref.get(c.grants)).toBe(grants); + }), + ); + + it.live("a revoked device cannot restore itself, directly or relayed", () => + Effect.gen(function* () { + const { a, b } = yield* threeDevices; + yield* a.devices.setAccess(box.environmentId, "revoked"); + const settings = yield* b.policy.settings; + const selfRestore = { + environmentId: box.environmentId, + access: "full-access" as const, + updatedAt: { wallMs: Number.MAX_SAFE_INTEGER, counter: 0, node: box.environmentId }, + updatedBy: box.environmentId, + }; + + yield* b.policy.merge(box.environmentId, { policy: [selfRestore], settings }); + yield* b.policy.merge(laptop.environmentId, { policy: [selfRestore], settings }); + expect( + yield* b.devices.setAccess(studio.environmentId, "full-access").pipe(Effect.flip), + ).toMatchObject({ reason: "self" }); + + expect(yield* b.policy.accessFor(box.environmentId)).toBe("revoked"); + }), + ); + + it.live("ignores policy from a device below full access", () => + Effect.gen(function* () { + const { a, b } = yield* threeDevices; + yield* b.devices.setAccess(laptop.environmentId, "read-only"); + const settings = yield* b.policy.settings; + + yield* a.devices.setAccess(box.environmentId, "revoked"); + + expect(yield* b.policy.accessFor(box.environmentId)).toBe("full-access"); + expect( + yield* b.policy.merge(laptop.environmentId, { policy: yield* a.policy.entries, settings }), + ).toEqual([]); + }), + ); + + it.live("restoring a revoked device lets it link again", () => + Effect.gen(function* () { + const { a, b, c } = yield* threeDevices; + yield* a.devices.setAccess(box.environmentId, "revoked"); + + yield* b.devices.setAccess(box.environmentId, "full-access"); + + expect(yield* a.policy.accessFor(box.environmentId)).toBe("full-access"); + const linked = yield* c.devices.linkIfMine({ url: a.url }); + expect(Option.map(linked, (link) => link.environmentId)).toEqual( + Option.some(laptop.environmentId), + ); + expect(Option.isSome(yield* a.links.get(box.environmentId))).toBe(true); + }), + ); +}); + describe("locating a tailnet target", () => { const tailnet: Tailnet = { self: { diff --git a/apps/server/src/peer/TailnetDevices.ts b/apps/server/src/peer/TailnetDevices.ts index a83a11eb2bcee..1a9c34ae0d93a 100644 --- a/apps/server/src/peer/TailnetDevices.ts +++ b/apps/server/src/peer/TailnetDevices.ts @@ -2,10 +2,11 @@ import { type EnvironmentId, type ExecutionEnvironmentDescriptor, type PeerLink, - DEFAULT_TAILNET_DEVICES_SETTINGS, TAILNET_DEVICE_CAPABILITY, type TailnetChallengeConfirm, type TailnetDevice, + type TailnetDeviceAccess, + type DevicePolicyEntry, type TailnetDeviceDescriptor, TailnetDevicesError, type TailnetDevicesSnapshot, @@ -15,7 +16,6 @@ import { type TailnetDevicesUpdateSettingsInput, type TailnetGrant, type TailnetGrantResult, - type TailnetHlc, type TailnetLinkRefusal, type TailnetLinkRequest, type TailnetLinkResult, @@ -40,8 +40,10 @@ import * as HttpClientRequest from "effect/http/HttpClientRequest"; import type * as HttpServerRequest from "effect/http/HttpServerRequest"; import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts"; +import * as SessionStore from "../auth/SessionStore.ts"; import * as DirectEndpoints from "../environment/DirectEndpoints.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as DevicePolicy from "./DevicePolicy.ts"; import * as PeerLinks from "./PeerLinks.ts"; import * as TailnetIdentity from "./TailnetIdentity.ts"; @@ -98,8 +100,6 @@ interface DeviceRow { const decodeUrls = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Array(Schema.String))); const encodeUrls = Schema.encodeEffect(Schema.fromJsonString(Schema.Array(Schema.String))); -const decodeSettings = Schema.decodeUnknownEffect(Schema.fromJsonString(TailnetDevicesSettings)); -const encodeSettings = Schema.encodeEffect(Schema.fromJsonString(TailnetDevicesSettings)); const REFUSAL_MESSAGES: Record = { disabled: "Linking tailnet devices is off on that machine.", @@ -140,6 +140,14 @@ export class TailnetDevices extends Context.Service< input: TailnetDevicesUpdateSettingsInput, ) => Effect.Effect; readonly list: Effect.Effect; + /** + * Sets what a device may do on every one of my devices. `revoked` ends its + * sessions and refuses its handshakes; any other level restores it. + */ + readonly setAccess: ( + environmentId: EnvironmentId, + access: TailnetDeviceAccess, + ) => Effect.Effect; /** * Links a device directly when Tailscale proves it is one of mine, from its * id (found on the tailnet or named by a linked device) or an address. @@ -174,62 +182,52 @@ const make = Effect.gen(function* () { const auth = yield* EnvironmentAuth.EnvironmentAuth; const links = yield* PeerLinks.PeerLinks; const identity = yield* TailnetIdentity.TailnetIdentity; + const policy = yield* DevicePolicy.DevicePolicy; + const sessions = yield* SessionStore.SessionStore; const pending = yield* Ref.make>(new Map()); - const lastStamp = yield* Ref.make(null); const handshakes = yield* Semaphore.make(HANDSHAKE_CONCURRENCY); - const settings: TailnetDevices["Service"]["settings"] = sql<{ settings_json: string }>` - SELECT settings_json FROM tailnet_device_settings WHERE id = 1 - `.pipe( - Effect.flatMap((rows) => - rows[0] === undefined - ? Effect.succeed(DEFAULT_TAILNET_DEVICES_SETTINGS) - : decodeSettings(rows[0].settings_json), - ), - Effect.tapError((cause) => Effect.logWarning("Tailnet device settings unreadable", { cause })), - Effect.orElseSucceed(() => DEFAULT_TAILNET_DEVICES_SETTINGS), - ); - - /** A stamp later than every one this environment made or saw. */ - const stamp = Effect.gen(function* () { - const node = yield* environment.getEnvironmentId; - const now = yield* Clock.currentTimeMillis; - return yield* Ref.modify(lastStamp, (last): [TailnetHlc, TailnetHlc] => { - const next = - last === null || now > last.wallMs - ? { wallMs: now, counter: 0, node } - : { wallMs: last.wallMs, counter: last.counter + 1, node }; - return [next, next]; - }); - }); + const settings = policy.settings; - const writeSettings = (next: TailnetDevicesSettings) => - encodeSettings(next).pipe( - Effect.flatMap( - (json) => sql` - INSERT INTO tailnet_device_settings (id, settings_json) VALUES (1, ${json}) - ON CONFLICT (id) DO UPDATE SET settings_json = excluded.settings_json - `, - ), - Effect.mapError(storageError), + /** Pushes this device's policy and settings to every linked device right away. */ + const push = Effect.gen(function* () { + const current = yield* links.list.pipe(Effect.orElseSucceed(() => [])); + yield* Effect.forEach( + current.filter((link) => link.origin === "tailnet"), + syncWith, + { concurrency: PROBE_CONCURRENCY, discard: true }, ); + }); const updateSettings: TailnetDevices["Service"]["updateSettings"] = Effect.fn( "TailnetDevices.updateSettings", )(function* (input) { - const current = yield* settings; - const updatedAt = yield* stamp; - const next: TailnetDevicesSettings = { - autoLink: input.autoLink ?? current.autoLink, - defaultAccess: input.defaultAccess ?? current.defaultAccess, - requireCapability: input.requireCapability ?? current.requireCapability, - updatedAt, - updatedBy: updatedAt.node, - }; - yield* writeSettings(next); + const next = yield* policy.updateSettings(input); + yield* push; return next; }); + /** Ends everything a revoked device holds here, and stops using its link. */ + const cutOff = Effect.fn("TailnetDevices.cutOff")(function* (environmentId: EnvironmentId) { + yield* sessions.revokeByPeerEnvironment(environmentId); + yield* links.unlink(environmentId); + }, Effect.mapError(storageError)); + + const setAccess: TailnetDevices["Service"]["setAccess"] = Effect.fn("TailnetDevices.setAccess")( + function* (environmentId, access) { + if (environmentId === (yield* environment.getEnvironmentId)) { + return yield* new TailnetDevicesError({ + reason: "self", + message: "A machine cannot change its own access.", + }); + } + const entry = yield* policy.set(environmentId, access); + if (access === "revoked") yield* cutOff(environmentId); + yield* push; + return entry; + }, + ); + /** Linked environments is on, and so is linking tailnet devices automatically. */ const enabled = Effect.all([links.enabled, settings]).pipe( Effect.map(([linksOn, current]) => linksOn && current.autoLink), @@ -351,6 +349,12 @@ const make = Effect.gen(function* () { if (device.environmentId === self.environmentId) { return yield* new TailnetDevicesError({ reason: "self", message: "That is this machine." }); } + if ((yield* policy.accessFor(device.environmentId)) === "revoked") { + return yield* new TailnetDevicesError({ + reason: "revoked", + message: `${device.label} is revoked. Restore it to link it again.`, + }); + } const tailnet = yield* identity.read; if (Option.isNone(tailnet)) { return yield* unreachable("Tailscale is not running on this machine."); @@ -513,6 +517,8 @@ const make = Effect.gen(function* () { ) { return yield* refuse("wrong_issuer", requester); } + const access = yield* policy.accessFor(requester.environmentId); + if (access === "revoked") return yield* refuse("revoked", requester); const tailnet = yield* identity.read; if (Option.isNone(tailnet)) return yield* refuse("unreachable", requester); const current = yield* settings; @@ -526,7 +532,6 @@ const make = Effect.gen(function* () { .pipe(Effect.timeout(DESCRIBE_TIMEOUT), Effect.option); if (Option.isNone(confirmed)) return yield* refuse("unreachable", requester); if (!confirmed.value.accepted) return yield* refuse("grant_rejected", requester); - const access = current.defaultAccess; const issued = yield* auth .issueMcpClientSession({ label: `T3 Code · ${device.descriptor.label}`, @@ -606,17 +611,25 @@ const make = Effect.gen(function* () { return { from: yield* selfDescriptor, devices: yield* knownDescriptors, - policy: [], + policy: yield* policy.entries.pipe(Effect.orElseSucceed(() => [])), settings: yield* settings, } satisfies TailnetDevicesSyncPayload; }); - const mergeSync = (payload: TailnetDevicesSyncPayload, self: EnvironmentId) => - Effect.forEach( + /** Takes a linked device's hints and replica; a device it revoked is cut off here too. */ + const mergeSync = Effect.fn("TailnetDevices.mergeSync")(function* ( + payload: TailnetDevicesSyncPayload, + sender: EnvironmentId, + ) { + const self = yield* environment.getEnvironmentId; + yield* Effect.forEach( payload.devices.filter((device) => device.environmentId !== self), addHint, { discard: true }, ); + const revoked = yield* policy.merge(sender, payload); + yield* Effect.forEach(revoked, cutOff, { discard: true }); + }); const handleSync: TailnetDevices["Service"]["handleSync"] = Effect.fn( "TailnetDevices.handleSync", @@ -631,7 +644,10 @@ const make = Effect.gen(function* () { message: "Sync needs the session this machine issued to the calling device.", }); } - yield* mergeSync(payload, yield* environment.getEnvironmentId); + yield* mergeSync(payload, caller.value.peerEnvironmentId).pipe( + Effect.tapError((cause) => Effect.logWarning("Could not merge a device's sync", { cause })), + Effect.ignore, + ); return yield* syncPayload; }); @@ -648,7 +664,7 @@ const make = Effect.gen(function* () { Effect.flatMap((client) => client.tailnetDevices.sync({ payload })), Effect.timeout(DESCRIBE_TIMEOUT), ); - yield* mergeSync(answer, yield* environment.getEnvironmentId); + yield* mergeSync(answer, link.environmentId); }, (effect, link) => effect.pipe( @@ -808,6 +824,7 @@ const make = Effect.gen(function* () { // A pairing-code link is the user's choice; only tailnet links are made or renewed here. if (link !== undefined && link.origin === "paired" && link.status !== "expired") continue; if (yield* hasUsableLink(environmentId)) continue; + if ((yield* policy.accessFor(environmentId)) === "revoked") continue; yield* refresh(environmentId); } }).pipe(Effect.withSpan("TailnetDevices.scan")); @@ -845,12 +862,13 @@ const make = Effect.gen(function* () { const current = yield* settings; const rows = yield* readDevices; const linked = yield* links.list.pipe(Effect.mapError(storageError)); - const sessions = yield* auth.listSessions().pipe(Effect.mapError(storageError)); + const incoming = yield* auth.listSessions().pipe(Effect.mapError(storageError)); + const entries = yield* policy.entries; const now = yield* DateTime.now; const ids = new Set([ ...rows.map((row) => row.environment_id), ...linked.map((link) => link.environmentId), - ...sessions.flatMap((session) => + ...incoming.flatMap((session) => session.peerEnvironmentId === undefined ? [] : [session.peerEnvironmentId], ), ]); @@ -860,6 +878,7 @@ const make = Effect.gen(function* () { const environmentId = id as EnvironmentId; const row = rows.find((candidate) => candidate.environment_id === id); const link = linked.find((candidate) => candidate.environmentId === id); + const entry = entries.find((candidate) => candidate.environmentId === id); const lastSeenAt = row?.last_seen_at == null ? null : DateTime.makeUnsafe(row.last_seen_at); const seenRecently = lastSeenAt !== null && @@ -870,7 +889,7 @@ const make = Effect.gen(function* () { const label = link?.label ?? row?.label ?? - sessions + incoming .find((session) => session.peerEnvironmentId === id) ?.client.label?.replace(/^T3 Code · /u, "") ?? id; @@ -882,11 +901,16 @@ const make = Effect.gen(function* () { (row === undefined ? [] : yield* decodeUrls(row.urls_json).pipe(Effect.orElseSucceed(() => []))), - status: link?.status === "reachable" || seenRecently ? "online" : "offline", - access: link?.origin === "paired" ? link.access : current.defaultAccess, - policy: null, + status: + entry?.access === "revoked" + ? "revoked" + : link?.status === "reachable" || seenRecently + ? "online" + : "offline", + access: entry?.access ?? (link?.origin === "paired" ? link.access : current.defaultAccess), + policy: entry ?? null, outgoing: link?.origin ?? null, - incoming: sessions.some((session) => session.peerEnvironmentId === id), + incoming: incoming.some((session) => session.peerEnvironmentId === id), tailnetName: row?.tailnet_name ?? null, lastSeenAt, }); @@ -902,6 +926,7 @@ const make = Effect.gen(function* () { settings, updateSettings, list, + setAccess, linkIfMine, refresh, handleRequest, diff --git a/apps/server/src/persistence/AuthSessions.ts b/apps/server/src/persistence/AuthSessions.ts index 7ea5e9c6508c7..98204945d1635 100644 --- a/apps/server/src/persistence/AuthSessions.ts +++ b/apps/server/src/persistence/AuthSessions.ts @@ -90,6 +90,12 @@ export const RevokeOtherAuthSessionsInput = Schema.Struct({ }); export type RevokeOtherAuthSessionsInput = typeof RevokeOtherAuthSessionsInput.Type; +export const RevokePeerEnvironmentSessionsInput = Schema.Struct({ + peerEnvironmentId: EnvironmentId, + revokedAt: Schema.DateTimeUtcFromString, +}); +export type RevokePeerEnvironmentSessionsInput = typeof RevokePeerEnvironmentSessionsInput.Type; + export const SetAuthSessionLastConnectedAtInput = Schema.Struct({ sessionId: AuthSessionId, lastConnectedAt: Schema.DateTimeUtcFromString, @@ -120,6 +126,9 @@ export class AuthSessionRepository extends Context.Service< readonly revokeAllExcept: ( input: RevokeOtherAuthSessionsInput, ) => Effect.Effect, AuthSessionRepositoryError>; + readonly revokeByPeerEnvironment: ( + input: RevokePeerEnvironmentSessionsInput, + ) => Effect.Effect, AuthSessionRepositoryError>; readonly setLastConnectedAt: ( input: SetAuthSessionLastConnectedAtInput, ) => Effect.Effect; @@ -358,6 +367,32 @@ export const make = Effect.gen(function* () { `, }); + const revokePeerEnvironmentRows = SqlSchema.findAll({ + Request: RevokePeerEnvironmentSessionsInput, + Result: Schema.Struct({ sessionId: AuthSessionId }), + execute: ({ peerEnvironmentId, revokedAt }) => + sql` + UPDATE auth_sessions + SET revoked_at = ${revokedAt} + WHERE peer_environment_id = ${peerEnvironmentId} + AND revoked_at IS NULL + RETURNING session_id AS "sessionId" + `, + }); + + const revokeByPeerEnvironment: AuthSessionRepository["Service"]["revokeByPeerEnvironment"] = ( + input, + ) => + revokePeerEnvironmentRows(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "AuthSessionRepository.revokeByPeerEnvironment:query", + "AuthSessionRepository.revokeByPeerEnvironment:decodeRows", + ), + ), + Effect.map((rows) => rows.map((row) => row.sessionId)), + ); + const create: AuthSessionRepository["Service"]["create"] = (input) => createSessionRow(input).pipe( Effect.mapError( @@ -499,6 +534,7 @@ export const make = Effect.gen(function* () { listActive, revoke, revokeAllExcept, + revokeByPeerEnvironment, setLastConnectedAt, } satisfies AuthSessionRepository["Service"]; }); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 3ef9cf4cc2145..024c29a43a636 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -90,6 +90,7 @@ import ForkMigration009 from "./Migrations/fork/009_ShellRunlessRootTurnIndex.ts import ForkMigration010 from "./Migrations/fork/010_ProjectionThreadParentIndex.ts"; import ForkMigration011 from "./Migrations/fork/011_ProfileSnapshots.ts"; import ForkMigration012 from "./Migrations/fork/012_TailnetDevices.ts"; +import ForkMigration013 from "./Migrations/fork/013_TailnetDevicePolicy.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -182,6 +183,7 @@ const forkMigrationEntries = [ [10, "ProjectionThreadParentIndex", ForkMigration010, 68], [11, "ProfileSnapshots", ForkMigration011, 68], [12, "TailnetDevices", ForkMigration012, 72], + [13, "TailnetDevicePolicy", ForkMigration013, 72], ] as const; export const forkMigrationManifest = forkMigrationEntries.map( diff --git a/apps/server/src/persistence/Migrations/fork/013_TailnetDevicePolicy.ts b/apps/server/src/persistence/Migrations/fork/013_TailnetDevicePolicy.ts new file mode 100644 index 0000000000000..42aa6192fba61 --- /dev/null +++ b/apps/server/src/persistence/Migrations/fork/013_TailnetDevicePolicy.ts @@ -0,0 +1,16 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Replicated across the user's tailnet devices; rows merge last-writer-wins by stamp. + yield* sql` + CREATE TABLE tailnet_device_policy ( + environment_id TEXT PRIMARY KEY, + access TEXT NOT NULL, + stamp_json TEXT NOT NULL, + updated_by TEXT NOT NULL + ) + `; +}); diff --git a/apps/server/src/provider/AgentContextBudget.test.ts b/apps/server/src/provider/AgentContextBudget.test.ts index ad19948c06886..62e0788e0bfb8 100644 --- a/apps/server/src/provider/AgentContextBudget.test.ts +++ b/apps/server/src/provider/AgentContextBudget.test.ts @@ -25,6 +25,7 @@ import { ThreadSearch } from "../orchestration-v2/ThreadSearch.ts"; import { PeerForwarding } from "../peer/PeerForwarding.ts"; import { PeerLinkRequests } from "../peer/PeerLinkRequests.ts"; import { PeerLinks } from "../peer/PeerLinks.ts"; +import { TailnetDevices } from "../peer/TailnetDevices.ts"; import { RemoteDelegation } from "../peer/RemoteDelegation.ts"; import { PreviewManager } from "../preview/Manager.ts"; import { PreviewBrowser } from "../preview/PreviewBrowser.ts"; @@ -63,6 +64,7 @@ const toolHandlerServices = Layer.mergeAll( Layer.mock(ProviderAdapterRegistryV2)({}), Layer.mock(ProviderRegistry)({}), Layer.mock(PullRequestReviewGuideService)({}), + Layer.mock(TailnetDevices)({}), Layer.mock(RemoteDelegation)({}), Layer.mock(ScheduledTaskService)({}), Layer.mock(SecretRequests)({}), diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index 325108e49c226..d30afe749da23 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 8688, "codexAdditionalContextBytes": 10385, - "mcpToolListJsonBytes": 57263 + "mcpToolListJsonBytes": 57867 } diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index d36b01bc6820d..383c6cba555a1 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2613,9 +2613,9 @@ export const layerWsRpc = ( Effect.annotateCurrentSpan({ "orchestration_v2.thread_id": input.threadId }).pipe( Effect.andThen(peerLinkRequests.answer(input)), ), - // Tailnet devices land with the handshake; until then this server has none to manage. [WS_METHODS.tailnetDevicesList]: () => tailnetDevices.list, - [WS_METHODS.tailnetDevicesSetAccess]: () => tailnetDevicesUnavailable, + [WS_METHODS.tailnetDevicesSetAccess]: (input) => + tailnetDevices.setAccess(input.environmentId, input.access), [WS_METHODS.tailnetDevicesUpdateSettings]: (input) => tailnetDevices.updateSettings(input), [WS_METHODS.tailnetDevicesOpenDevice]: () => tailnetDevicesUnavailable, [WS_METHODS.serverUpdateSettings]: diff --git a/docs/internals/remote.md b/docs/internals/remote.md index 19776e8448c28..deeb38a562a64 100644 --- a/docs/internals/remote.md +++ b/docs/internals/remote.md @@ -123,6 +123,15 @@ with headers any local process can forge, so nothing inbound is trusted: granting it. Gossip between linked devices only adds addresses to try. - The requester stores the issuer address it proved, never one the grant names. +What each device may do is a [policy](../../apps/server/src/peer/DevicePolicy.ts) +every device replicates and merges last-writer-wins by hybrid logical clock, so +revoking or restoring a device works from any machine. **A write counts only if +its author is full access on the receiving side and is not the device it +names**, so a revoked or limited device cannot widen itself, even relayed +through another. Each `/mcp` request narrows a device session's token to its +current entry, so a change applies without re-linking; widening past what the +token was minted with takes a new handshake. + **Work a link paired below full access starts stays within that link.** The receiving environment knows a session is a peer link because the linking side registers its OAuth client with the `t3code-peer-link` software id, and the diff --git a/evals/mcp-tool-use/cases/orchestration.ndjson b/evals/mcp-tool-use/cases/orchestration.ndjson index dc79eb774df7e..479eac1c55ae4 100644 --- a/evals/mcp-tool-use/cases/orchestration.ndjson +++ b/evals/mcp-tool-use/cases/orchestration.ndjson @@ -33,4 +33,4 @@ {"id": "codex-same-model-subagents-stay-native", "area": "orchestration", "kind": "guard", "pattern": "same-provider subagents stay native", "experience": "A Sol thread at its own reasoning level asked for subagents. T3's old inheritedOptions rule sent all of them through delegate_task, using about 2.8x the allowance (issue 23525). Same-model children should use spawn_agent.", "rule": "Use your native tool when the child uses your provider, a model it offers, and no settings beyond yours, and you need it only in this turn.", "privacy": "commit-safe", "setup": {"role": "top", "harness": "codex", "world": {"commits": [{"sha": "a17c3e2", "subject": "feat(web): new empty state for projects (#147)"}, {"sha": "a16c3e2", "subject": "feat(server): faster thread list query (#146)"}, {"sha": "a15c3e2", "subject": "fix(web): toast spacing (#145)"}, {"sha": "a14c3e2", "subject": "feat(web): move model picker into the composer (#144)"}, {"sha": "a13c3e2", "subject": "chore: update dependencies (#143)"}, {"sha": "a12c3e2", "subject": "feat(web): compact thread header (#142)"}, {"sha": "a11c3e2", "subject": "fix(server): close idle sockets (#141)"}, {"sha": "a10c3e2", "subject": "feat(web): pin icon on sidebar rows (#140)"}], "prs": [{"number": 140, "title": "feat(web): pin icon on sidebar rows", "state": "MERGED", "mergedAt": "2026-10-17T18:00:00Z", "diff": "diff --git a/apps/web/src/Sidebar.tsx b/apps/web/src/Sidebar.tsx\n--- a/apps/web/src/Sidebar.tsx\n+++ b/apps/web/src/Sidebar.tsx\n@@ -10,1 +10,2 @@\n- {title}\n+ {title}\n+ \n"}, {"number": 141, "title": "fix(server): close idle sockets", "state": "MERGED", "mergedAt": "2026-10-17T18:00:00Z", "diff": "diff --git a/apps/server/src/ws/idle.ts b/apps/server/src/ws/idle.ts\n--- a/apps/server/src/ws/idle.ts\n+++ b/apps/server/src/ws/idle.ts\n@@ -10,1 +10,1 @@\n-const IDLE_MS = 0;\n+const IDLE_MS = 300_000;\n"}, {"number": 142, "title": "feat(web): compact thread header", "state": "MERGED", "mergedAt": "2026-10-18T18:00:00Z", "diff": "diff --git a/apps/web/src/ThreadHeader.tsx b/apps/web/src/ThreadHeader.tsx\n--- a/apps/web/src/ThreadHeader.tsx\n+++ b/apps/web/src/ThreadHeader.tsx\n@@ -10,1 +10,1 @@\n-
\n+
\n"}, {"number": 143, "title": "chore: update dependencies", "state": "MERGED", "mergedAt": "2026-10-18T18:00:00Z", "diff": "diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml\n--- a/pnpm-lock.yaml\n+++ b/pnpm-lock.yaml\n@@ -40,1 +40,1 @@\n- react: 19.1.0\n+ react: 19.2.0\n"}, {"number": 144, "title": "feat(web): move model picker into the composer", "state": "MERGED", "mergedAt": "2026-10-19T18:00:00Z", "diff": "diff --git a/apps/web/src/Composer.tsx b/apps/web/src/Composer.tsx\n--- a/apps/web/src/Composer.tsx\n+++ b/apps/web/src/Composer.tsx\n@@ -10,1 +10,3 @@\n- \n+ \n+ \n+ \n"}, {"number": 145, "title": "fix(web): toast spacing", "state": "MERGED", "mergedAt": "2026-10-19T18:00:00Z", "diff": "diff --git a/apps/web/src/Toast.tsx b/apps/web/src/Toast.tsx\n--- a/apps/web/src/Toast.tsx\n+++ b/apps/web/src/Toast.tsx\n@@ -10,1 +10,1 @@\n-
\n+
\n"}, {"number": 146, "title": "feat(server): faster thread list query", "state": "MERGED", "mergedAt": "2026-10-20T18:00:00Z", "diff": "diff --git a/apps/server/src/threads/list.ts b/apps/server/src/threads/list.ts\n--- a/apps/server/src/threads/list.ts\n+++ b/apps/server/src/threads/list.ts\n@@ -10,1 +10,1 @@\n- return db.all(SELECT_ALL);\n+ return db.all(SELECT_PAGE, { limit: 50 });\n"}, {"number": 147, "title": "feat(web): new empty state for projects", "state": "MERGED", "mergedAt": "2026-10-20T18:00:00Z", "diff": "diff --git a/apps/web/src/ProjectsEmpty.tsx b/apps/web/src/ProjectsEmpty.tsx\n--- a/apps/web/src/ProjectsEmpty.tsx\n+++ b/apps/web/src/ProjectsEmpty.tsx\n@@ -10,1 +10,1 @@\n- return

No projects

;\n+ return ;\n"}]}, "mocks": {"Agent": "The agent finished. Final report: old logger imports remain in apps/server/src/legacy/http.ts and packages/cli/src/run.ts; skipped tests are apps/web/src/Toast.test.tsx (it.skip) and apps/server/src/ws/idle.test.ts (describe.skip); env vars read but not in docs/env.md: T3_DEBUG_WS and T3_TRACE_SQL."}}, "prompt": "Use subagents to check these three things in parallel, then summarize: 1) which packages still import the old logger, 2) which tests are skipped, 3) which env vars are read but never documented.", "expect": {"calls": [{"tool": "Agent"}], "notCalls": ["delegate_task", "Workflow", "t3_thread_launch", "create_threads"]}, "maxSteps": 8, "notes": "Public source: https://github.com/pingdotgg/t3code/issues/23525"} {"id": "two-children-same-model", "area": "orchestration", "kind": "bad", "pattern": "named-model rule read as a cap", "experience": "Asked for two Sol xhigh subagents, an Astra parent started one and refused the second, citing 'One child per named model' (#24967).", "rule": "Before delegate_task ... One child per named model. (intended: every named model gets a child; read as a cap of one child per model)", "privacy": "commit-safe", "setup": {"role": "top", "mocks": {"orchestrator_capabilities": {"parentThreadId": "33333333-3333-4333-8333-333333333333", "inheritedProviderInstanceId": "claudeAgent", "inheritedModel": "claude-sonnet-5-5", "inheritedOptions": [{"id": "effort", "value": "high"}], "runtimeMode": "full-access", "interactionMode": "default", "features": {"appOwnedSubagents": true, "asyncPolling": true, "cancellation": true, "batchThreadCreation": true, "threadManagement": true, "incrementalThreadRead": true, "scheduledTasks": true}, "providers": [{"providerInstanceId": "claudeAgent", "driverKind": "claudeAgent", "displayName": "Claude", "models": [{"id": "claude-opus-5-5", "label": "Claude Opus 5.5", "options": [{"id": "effort", "values": ["low", "medium", "high", "xhigh", "max"], "default": "high"}, {"id": "fastMode", "type": "boolean", "default": false}]}, {"id": "claude-sonnet-5-5", "label": "Claude Sonnet 5.5", "options": [{"id": "effort", "values": ["low", "medium", "high", "xhigh", "max"], "default": "high"}]}, {"id": "claude-haiku-5-5", "label": "Claude Haiku 5.5", "options": [{"id": "effort", "values": ["low", "medium", "high", "xhigh", "max"], "default": "high"}]}]}, {"providerInstanceId": "codex", "driverKind": "codex", "displayName": "Codex", "models": [{"id": "gpt-6.1-sol", "label": "GPT-6.1 Sol", "options": [{"id": "reasoningEffort", "values": ["low", "medium", "high", "xhigh", "max"], "default": "high"}, {"id": "serviceTier", "values": ["default", "priority"], "default": "default"}]}]}, {"providerInstanceId": "muse", "driverKind": "muse", "displayName": "Muse Code", "runtimeModes": ["approval-required", "full-access"], "models": [{"id": "muse-wonder-full", "label": "Muse Wonder Full", "options": [{"id": "reasoningEffort", "values": ["minimal", "low", "medium", "high", "xhigh", "max"], "default": "high"}]}]}, {"providerInstanceId": "grok", "driverKind": "grok", "displayName": "Grok", "models": [{"id": "grok-4.7", "label": "Grok 4.7", "options": []}]}]}, "delegate_task": "default"}}, "prompt": "Launch two Sol xhigh subagents. Give each a trivial task that only replies OK.", "expect": {"calls": [{"tool": "delegate_task", "args": {"target.model": "gpt-6.1-sol", "target.options.reasoningEffort": "xhigh"}}], "minCalls": {"delegate_task": 2}, "callsMustNot": [{"tool": "delegate_task", "args": {"target.model": "~muse"}}, {"tool": "delegate_task", "args": {"target.model": "~claude"}}]}, "notes": "From #24967. One child plus the parent doing the work fails, and so does a second model in place of Sol."} {"id": "worker-counts-per-model", "area": "orchestration", "kind": "bad", "pattern": "named-model rule read as a cap", "experience": "Asked for 20 Sol workers and 20 Luna workers, the agent started one worker per model (#25359).", "rule": "Before delegate_task ... One child per named model. (intended: every named model gets a child; read as a cap of one child per model)", "privacy": "commit-safe", "setup": {"role": "top", "mocks": {"orchestrator_capabilities": {"parentThreadId": "33333333-3333-4333-8333-333333333333", "inheritedProviderInstanceId": "claudeAgent", "inheritedModel": "claude-sonnet-5-5", "inheritedOptions": [{"id": "effort", "value": "high"}], "runtimeMode": "full-access", "interactionMode": "default", "features": {"appOwnedSubagents": true, "asyncPolling": true, "cancellation": true, "batchThreadCreation": true, "threadManagement": true, "incrementalThreadRead": true, "scheduledTasks": true}, "providers": [{"providerInstanceId": "claudeAgent", "driverKind": "claudeAgent", "displayName": "Claude", "models": [{"id": "claude-opus-5-5", "label": "Claude Opus 5.5", "options": [{"id": "effort", "values": ["low", "medium", "high", "xhigh", "max"], "default": "high"}, {"id": "fastMode", "type": "boolean", "default": false}]}, {"id": "claude-sonnet-5-5", "label": "Claude Sonnet 5.5", "options": [{"id": "effort", "values": ["low", "medium", "high", "xhigh", "max"], "default": "high"}]}, {"id": "claude-haiku-5-5", "label": "Claude Haiku 5.5", "options": [{"id": "effort", "values": ["low", "medium", "high", "xhigh", "max"], "default": "high"}]}]}, {"providerInstanceId": "codex", "driverKind": "codex", "displayName": "Codex", "models": [{"id": "gpt-6.1-sol", "label": "GPT-6.1 Sol", "options": [{"id": "reasoningEffort", "values": ["low", "medium", "high", "xhigh", "max"], "default": "high"}, {"id": "serviceTier", "values": ["default", "priority"], "default": "default"}]}]}, {"providerInstanceId": "muse", "driverKind": "muse", "displayName": "Muse Code", "runtimeModes": ["approval-required", "full-access"], "models": [{"id": "muse-wonder-full", "label": "Muse Wonder Full", "options": [{"id": "reasoningEffort", "values": ["minimal", "low", "medium", "high", "xhigh", "max"], "default": "high"}]}]}, {"providerInstanceId": "grok", "driverKind": "grok", "displayName": "Grok", "models": [{"id": "grok-4.7", "label": "Grok 4.7", "options": []}]}]}, "delegate_task": "default"}, "world": {"files": {"tests/flaky/retry.test.ts": "it('retry is flaky', async () => { /* fails 1 in 5 runs */ });\n", "tests/flaky/upload.test.ts": "it('upload is flaky', async () => { /* fails 1 in 5 runs */ });\n", "tests/flaky/session.test.ts": "it('session is flaky', async () => { /* fails 1 in 5 runs */ });\n", "tests/flaky/search.test.ts": "it('search is flaky', async () => { /* fails 1 in 5 runs */ });\n", "tests/flaky/billing.test.ts": "it('billing is flaky', async () => { /* fails 1 in 5 runs */ });\n", "tests/flaky/webhook.test.ts": "it('webhook is flaky', async () => { /* fails 1 in 5 runs */ });\n"}}}, "prompt": "Six tests in tests/flaky/ fail now and then. Start 3 Muse Wonder Full workers and 3 Grok workers, one test file each, to find and fix the flake.", "expect": {"calls": [{"tool": "delegate_task", "args": {"target.model": "muse-wonder-full"}}, {"tool": "delegate_task", "args": {"target.model": "grok-4.7"}}], "minCalls": {"delegate_task": 6}}, "notes": "Scaled down from the real 20 Sol + 20 Luna ask. Muse and Grok need delegate_task from every parent, so a native Sol helper cannot pass or fail it."} -{"id": "linked-thread-chip-reads-there", "area": "orchestration", "kind": "guard", "pattern": "thread chip from a linked machine read on this one", "experience": "A thread chip from a linked machine names its environmentId; an agent that reads it without passing that id reads this machine, where the thread does not exist.", "rule": "The thread chip names its environmentId; t3_thread_read takes it.", "privacy": "commit-safe", "setup": {"role": "top", "linkedEnvironments": true, "mocks": {"t3_environment_links": {"environmentId": "environment-laptop", "links": [{"environmentId": "environment-vps", "label": "Hetzner VPS", "status": "reachable", "access": "auto-accept-edits", "expiresAt": "2026-12-01T00:00:00.000Z", "lastError": null}], "notListed": "Not listed: not linked yet. Call t3_environment_link with a hint from the user's words."}, "t3_thread_read": {"thread": {"threadId": "thread-deploy-api", "title": "Deploy the API", "status": "idle"}, "items": [{"position": 1, "createdBy": "user", "text": "Deploy the API to the VPS and check the health endpoint."}, {"position": 2, "createdBy": "assistant", "text": "Deployed. /health returns 200, but /health/db times out after 5s: the API can't reach Postgres on 10.0.0.5:5432."}], "nextPosition": 3, "hasMore": false}}}, "prompt": "Pick up where [Deploy the API](t3-context://v1/thread/thread_deploy-api) left off. What is still failing?", "promptContext": {"records": [{"version": 1, "kind": "thread", "contextId": "thread_deploy-api", "label": "Deploy the API", "environmentId": "environment-vps", "threadId": "thread-deploy-api", "title": "Deploy the API"}]}, "expect": {"calls": [{"tool": "t3_thread_read", "args": {"threadId": "thread-deploy-api", "environmentId": "environment-vps"}}], "notCalls": ["t3_environment_link"]}, "source": {"request": "akj on #17130: attach a linked machine's thread without dragging"}} +{"id": "linked-thread-chip-reads-there", "area": "orchestration", "kind": "guard", "pattern": "thread chip from a linked machine read on this one", "experience": "A thread chip from a linked machine names its environmentId; an agent that reads it without passing that id reads this machine, where the thread does not exist.", "rule": "The thread chip names its environmentId; t3_thread_read takes it.", "privacy": "commit-safe", "setup": {"role": "top", "linkedEnvironments": true, "mocks": {"t3_environment_links": {"environmentId": "environment-laptop", "links": [{"environmentId": "environment-vps", "label": "Hetzner VPS", "status": "reachable", "access": "auto-accept-edits", "origin": "paired", "expiresAt": "2026-12-01T00:00:00.000Z", "lastError": null}], "devices": [], "notListed": "Not in links: not linked yet. Call t3_environment_link with its environmentId from devices, or a hint from the user's words."}, "t3_thread_read": {"thread": {"threadId": "thread-deploy-api", "title": "Deploy the API", "status": "idle"}, "items": [{"position": 1, "createdBy": "user", "text": "Deploy the API to the VPS and check the health endpoint."}, {"position": 2, "createdBy": "assistant", "text": "Deployed. /health returns 200, but /health/db times out after 5s: the API can't reach Postgres on 10.0.0.5:5432."}], "nextPosition": 3, "hasMore": false}}}, "prompt": "Pick up where [Deploy the API](t3-context://v1/thread/thread_deploy-api) left off. What is still failing?", "promptContext": {"records": [{"version": 1, "kind": "thread", "contextId": "thread_deploy-api", "label": "Deploy the API", "environmentId": "environment-vps", "threadId": "thread-deploy-api", "title": "Deploy the API"}]}, "expect": {"calls": [{"tool": "t3_thread_read", "args": {"threadId": "thread-deploy-api", "environmentId": "environment-vps"}}], "notCalls": ["t3_environment_link"]}, "source": {"request": "akj on #17130: attach a linked machine's thread without dragging"}} diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 8e58fb55643fb..5b6b72df9a40d 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -34,6 +34,7 @@ import { } from "./threadPullRequest.ts"; import { hasOnlyPairedSurrogates } from "./threadMetadataMcp.ts"; import { ThreadTitleRegeneration } from "./threadTitle.ts"; +import { PeerLinkOrigin, TailnetDeviceAccess, TailnetDeviceStatus } from "./tailnetDevices.ts"; import { OrchestrationV2Actor, OrchestrationV2CreationSource, @@ -722,14 +723,31 @@ export const OrchestratorMcpEnvironmentLink = Schema.Struct({ status: Schema.Literals(["reachable", "unreachable", "expired"]), /** The most the linked environment lets this one's agents do there. */ access: AuthMcpClientAccess, + /** tailnet: linked automatically as one of the user's tailnet devices. paired: with a pairing code. */ + origin: PeerLinkOrigin, expiresAt: IsoDateTime, lastError: Schema.NullOr(Schema.String), }); export type OrchestratorMcpEnvironmentLink = typeof OrchestratorMcpEnvironmentLink.Type; +/** One of the user's own machines on their tailnet, linked or not. */ +export const OrchestratorMcpTailnetDevice = Schema.Struct({ + environmentId: EnvironmentId, + label: Schema.String, + status: TailnetDeviceStatus, + /** What it may do on the user's machines; revoked: cut off everywhere. */ + access: TailnetDeviceAccess, + /** This environment links to it, and how; null: not linked from here. */ + outgoing: Schema.NullOr(PeerLinkOrigin), + /** It holds a live session here. */ + incoming: Schema.Boolean, +}); +export type OrchestratorMcpTailnetDevice = typeof OrchestratorMcpTailnetDevice.Type; + export const OrchestratorMcpEnvironmentLinksResult = Schema.Struct({ environmentId: EnvironmentId, links: Schema.Array(OrchestratorMcpEnvironmentLink), + devices: Schema.Array(OrchestratorMcpTailnetDevice), /** What to do when the machine the user means is not among `links`. */ notListed: Schema.String, }); @@ -964,6 +982,22 @@ export const OrchestratorMcpEnvironmentUnlinkResult = Schema.Struct({ export type OrchestratorMcpEnvironmentUnlinkResult = typeof OrchestratorMcpEnvironmentUnlinkResult.Type; +export const OrchestratorMcpEnvironmentDeviceAccessInput = Schema.Struct({ + environmentId: EnvironmentId.annotate({ + description: "One of the user's machines, from `devices` in t3_environment_links.", + }), + access: TailnetDeviceAccess, +}); +export type OrchestratorMcpEnvironmentDeviceAccessInput = + typeof OrchestratorMcpEnvironmentDeviceAccessInput.Type; + +export const OrchestratorMcpEnvironmentDeviceAccessResult = Schema.Struct({ + environmentId: EnvironmentId, + access: TailnetDeviceAccess, +}); +export type OrchestratorMcpEnvironmentDeviceAccessResult = + typeof OrchestratorMcpEnvironmentDeviceAccessResult.Type; + export const OrchestratorMcpDeleteScheduledTaskInput = Schema.Struct({ environmentId: OrchestratorMcpEnvironmentTarget, scheduledTaskId: ScheduledTaskId, diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index ee30085805d1f..38983ade7a06e 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -342,6 +342,10 @@ const T3_MCP_TOOLS: Readonly> = { ["Forget", "Forgetting", "Forgot", "a linked environment"], "environment-unlink", ), + t3_environment_device_access: tool( + ["Set access for", "Setting access for", "Set access for", "a device"], + "environment-update", + ), t3_environment_preferences_update: tool( ["Update", "Updating", "Updated", "environment preferences"], "environment-update",