From 163b7a684de0ec10eb9b7a3501e76b120f29f156 Mon Sep 17 00:00:00 2001 From: Andrew Brookins Date: Mon, 4 May 2026 09:36:48 -0700 Subject: [PATCH 1/3] Fix ownerless MCP token minting --- cmd/afs/auth_commands_test.go | 2 + cmd/afs/file_commands_test.go | 3 +- internal/controlplane/http_test.go | 99 +++++++++++++++++++++++++++++ internal/controlplane/mcp_tokens.go | 6 ++ sdk/python/tests/test_client.py | 76 +++++++++++++++++++++- 5 files changed, 184 insertions(+), 2 deletions(-) diff --git a/cmd/afs/auth_commands_test.go b/cmd/afs/auth_commands_test.go index 89e77c0..1aa8c69 100644 --- a/cmd/afs/auth_commands_test.go +++ b/cmd/afs/auth_commands_test.go @@ -254,6 +254,8 @@ func TestCmdAuthStatusShowsSignedInCloudState(t *testing.T) { func TestCmdStatusShowsSignedInCloudState(t *testing.T) { t.Helper() + withTempHome(t) + cfg := defaultConfig() cfg.ProductMode = productModeCloud cfg.URL = "https://afs.example.com" diff --git a/cmd/afs/file_commands_test.go b/cmd/afs/file_commands_test.go index 68551d5..de67ded 100644 --- a/cmd/afs/file_commands_test.go +++ b/cmd/afs/file_commands_test.go @@ -195,6 +195,7 @@ func TestCmdFSHistoryReportsExistingFileWithoutRecordedHistory(t *testing.T) { cfg := defaultConfig() cfg.RedisAddr = mr.Addr() cfg.MountBackend = mountBackendNone + cfg.CurrentWorkspace = "repo" saveTempConfig(t, cfg) loadedCfg, store, closeStore, err := openAFSStore(context.Background()) @@ -214,7 +215,7 @@ func TestCmdFSHistoryReportsExistingFileWithoutRecordedHistory(t *testing.T) { } output, err := captureStdout(t, func() error { - return cmdFS([]string{"fs", "-w", "repo", "history", "helloworld.txt"}) + return cmdFS([]string{"fs", "history", "helloworld.txt"}) }) if err != nil { t.Fatalf("cmdFS(history) returned error: %v", err) diff --git a/internal/controlplane/http_test.go b/internal/controlplane/http_test.go index ae96f8a..9c2b104 100644 --- a/internal/controlplane/http_test.go +++ b/internal/controlplane/http_test.go @@ -1141,6 +1141,105 @@ func TestMCPTokenFlowWorksWithoutAuth(t *testing.T) { } } +func TestHostedControlPlaneTokenCanIssueWorkspaceTokenWithoutAuth(t *testing.T) { + t.Helper() + + manager, _ := newTestManager(t) + server := httptest.NewServer(NewHandler(manager, "*")) + defer server.Close() + + createReq, err := http.NewRequest( + http.MethodPost, + server.URL+"/v1/mcp-tokens", + strings.NewReader(`{"name":"Control plane token"}`), + ) + if err != nil { + t.Fatalf("NewRequest(create control-plane token) returned error: %v", err) + } + createReq.Header.Set("Content-Type", "application/json") + createResp, err := http.DefaultClient.Do(createReq) + if err != nil { + t.Fatalf("POST control-plane mcp token returned error: %v", err) + } + defer createResp.Body.Close() + if createResp.StatusCode != http.StatusCreated { + body, _ := io.ReadAll(createResp.Body) + t.Fatalf("POST control-plane mcp token status = %d, want %d, body=%s", createResp.StatusCode, http.StatusCreated, body) + } + + var controlPlaneToken mcpAccessTokenResponse + if err := json.NewDecoder(createResp.Body).Decode(&controlPlaneToken); err != nil { + t.Fatalf("Decode(control-plane mcp token) returned error: %v", err) + } + if controlPlaneToken.Token == "" { + t.Fatal("expected created control-plane mcp token secret") + } + if controlPlaneToken.Scope != mcpScopeControlPlane { + t.Fatalf("control-plane token scope = %q, want %q", controlPlaneToken.Scope, mcpScopeControlPlane) + } + + callBody := `{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"mcp_token_issue","arguments":{"workspace":"repo","name":"Mounted FS","profile":"workspace-rw"}}}` + callReq, err := http.NewRequest(http.MethodPost, server.URL+"/mcp", strings.NewReader(callBody)) + if err != nil { + t.Fatalf("NewRequest(mcp_token_issue) returned error: %v", err) + } + callReq.Header.Set("Content-Type", "application/json") + callReq.Header.Set("Authorization", "Bearer "+controlPlaneToken.Token) + callReq.Header.Set("Accept", "application/json, text/event-stream") + callResp, err := http.DefaultClient.Do(callReq) + if err != nil { + t.Fatalf("POST /mcp mcp_token_issue returned error: %v", err) + } + defer callResp.Body.Close() + if callResp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(callResp.Body) + t.Fatalf("POST /mcp mcp_token_issue status = %d, want %d, body=%s", callResp.StatusCode, http.StatusOK, body) + } + + var issuePayload struct { + Result struct { + StructuredContent map[string]any `json:"structuredContent"` + } `json:"result"` + } + if err := json.NewDecoder(callResp.Body).Decode(&issuePayload); err != nil { + t.Fatalf("Decode(mcp_token_issue response) returned error: %v", err) + } + issuedToken, _ := issuePayload.Result.StructuredContent["token"].(string) + if issuedToken == "" { + t.Fatalf("mcp_token_issue token = %#v, want non-empty", issuePayload.Result.StructuredContent["token"]) + } + if got := issuePayload.Result.StructuredContent["workspace"]; got != "repo" { + t.Fatalf("mcp_token_issue workspace = %#v, want %q", got, "repo") + } + if got := issuePayload.Result.StructuredContent["profile"]; got != MCPProfileWorkspaceRW { + t.Fatalf("mcp_token_issue profile = %#v, want %q", got, MCPProfileWorkspaceRW) + } + if got, _ := issuePayload.Result.StructuredContent["scope"].(string); !strings.HasPrefix(got, mcpScopeWorkspacePrefix) { + t.Fatalf("mcp_token_issue scope = %#v, want workspace scope", issuePayload.Result.StructuredContent["scope"]) + } + + toolsReq, err := http.NewRequest( + http.MethodPost, + server.URL+"/mcp", + strings.NewReader(`{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}`), + ) + if err != nil { + t.Fatalf("NewRequest(workspace-token tools/list) returned error: %v", err) + } + toolsReq.Header.Set("Content-Type", "application/json") + toolsReq.Header.Set("Authorization", "Bearer "+issuedToken) + toolsReq.Header.Set("Accept", "application/json, text/event-stream") + toolsResp, err := http.DefaultClient.Do(toolsReq) + if err != nil { + t.Fatalf("POST /mcp workspace-token tools/list returned error: %v", err) + } + defer toolsResp.Body.Close() + if toolsResp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(toolsResp.Body) + t.Fatalf("POST /mcp workspace-token tools/list status = %d, want %d, body=%s", toolsResp.StatusCode, http.StatusOK, body) + } +} + func TestHTTPResolvedOnboardingTokenExchange(t *testing.T) { t.Helper() diff --git a/internal/controlplane/mcp_tokens.go b/internal/controlplane/mcp_tokens.go index ba2451e..717da25 100644 --- a/internal/controlplane/mcp_tokens.go +++ b/internal/controlplane/mcp_tokens.go @@ -463,6 +463,12 @@ func (m *DatabaseManager) requireOwnedSubject(ctx context.Context) (string, stri return "", "", nil } if strings.TrimSpace(identity.Subject) == "" { + // Self-managed no-auth installs can reach this path with an ownerless + // control-plane MCP token. Those tokens are intentionally allowed to mint + // workspace-scoped MCP tokens within the self-managed policy surface. + if strings.TrimSpace(identity.Provider) == "mcp-token" && isControlPlaneScope(identity.Scope) { + return "", "", nil + } if strings.TrimSpace(identity.Provider) == "" || strings.TrimSpace(identity.Provider) == string(AuthModeNone) { return "", "", nil } diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index 7391214..e7c1fef 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -1,6 +1,7 @@ import unittest +from unittest.mock import patch -from redis_afs.client import MCPHttpClient, AFSError, MountedFS, _MountedWorkspace, _normalize_mcp_endpoint +from redis_afs.client import AFSError, FSClient, MCPHttpClient, MountedFS, _MountedWorkspace, _normalize_mcp_endpoint class FakeMCP: @@ -34,6 +35,64 @@ def call_tool(self, name, arguments=None): raise AssertionError(f"unexpected tool {name}") +class FakeControlPlane: + def __init__(self): + self.issued = [] + self.timeout = 30.0 + self.endpoint = "https://afs.example/mcp" + + def call_tool(self, name, arguments=None): + arguments = arguments or {} + if name != "mcp_token_issue": + raise AssertionError(f"unexpected tool {name}") + token = f"workspace-token-{arguments['workspace']}" + self.issued.append({"name": name, "arguments": dict(arguments), "token": token}) + return { + "token": token, + "url": "https://afs.example/mcp", + "workspace": arguments["workspace"], + "profile": arguments["profile"], + "scope": f"workspace:{arguments['workspace']}", + } + + +class FakeMountedMCPHttpClient: + files_by_token = {} + + def __init__(self, *, api_key, base_url=None, timeout=30.0, headers=None): + self.api_key = api_key + self.endpoint = base_url or "https://afs.example/mcp" + self.timeout = timeout + self.headers = dict(headers or {}) + + def call_tool(self, name, arguments=None): + arguments = arguments or {} + files = self.files_by_token.setdefault(self.api_key, {}) + if name == "file_write": + files[arguments["path"]] = arguments["content"] + return {"path": arguments["path"], "operation": "write"} + if name == "file_read": + return { + "path": arguments["path"], + "kind": "file", + "content": files.get(arguments["path"], ""), + } + if name == "file_list": + path = arguments.get("path", "/") + entries = [] + for file_path in sorted(files): + if path == "/" and "/" not in file_path.strip("/"): + entries.append({"path": file_path, "name": file_path.strip("/"), "kind": "file"}) + elif file_path.startswith(path.rstrip("/") + "/"): + remainder = file_path[len(path.rstrip("/")) + 1 :] + if "/" not in remainder: + entries.append({"path": file_path, "name": remainder, "kind": "file"}) + return {"entries": entries} + if name == "checkpoint_create": + return {"workspace": "workspace", "checkpoint": arguments.get("checkpoint") or "auto", "created": True} + raise AssertionError(f"unexpected tool {name}") + + class MountedFSTest(unittest.TestCase): def test_single_workspace_paths_are_workspace_relative(self): fake = FakeMCP() @@ -68,6 +127,21 @@ def test_maps_absolute_workspace_paths_after_materialization(self): self.assertIn(root, mapped) self.assertNotEqual(mapped, "cat /foobar/README.md") + def test_fs_mount_issues_workspace_token_and_reads_and_writes_files(self): + control_plane = FakeControlPlane() + + with patch("redis_afs.client.MCPHttpClient", FakeMountedMCPHttpClient): + fs = FSClient(control_plane).mount(workspaces=[{"name": "repo"}], mode="rw", token_name="Mounted FS") + self.addCleanup(fs.close) + + fs.write_file("/repo/README.md", "hello from mounted fs") + + self.assertEqual(fs.read_file("/repo/README.md"), "hello from mounted fs") + self.assertEqual(fs.workspace_names, ["repo"]) + self.assertEqual(control_plane.issued[0]["arguments"]["workspace"], "repo") + self.assertEqual(control_plane.issued[0]["arguments"]["profile"], "workspace-rw") + self.assertEqual(control_plane.issued[0]["arguments"]["name"], "Mounted FS") + class EndpointTest(unittest.TestCase): def test_normalizes_mcp_endpoint(self): From 7409620e117dd6045b851f6269fc9b31c905d11a Mon Sep 17 00:00:00 2001 From: Andrew Brookins Date: Mon, 4 May 2026 20:09:40 -0700 Subject: [PATCH 2/3] Restore workspace-first change routes (#9) --- internal/controlplane/database_manager.go | 36 +++++++- internal/controlplane/http.go | 58 ++++++++++++ internal/controlplane/http_test.go | 102 ++++++++++++++++++++++ 3 files changed, 195 insertions(+), 1 deletion(-) diff --git a/internal/controlplane/database_manager.go b/internal/controlplane/database_manager.go index ce2521c..05d1ab3 100644 --- a/internal/controlplane/database_manager.go +++ b/internal/controlplane/database_manager.go @@ -1128,7 +1128,25 @@ func (m *DatabaseManager) ListChangelog(ctx context.Context, databaseID, workspa if err != nil { return ChangelogListResponse{}, err } - return service.ListChangelog(ctx, route.WorkspaceID, req) + response, err := service.ListChangelog(ctx, route.WorkspaceID, req) + if err != nil { + return ChangelogListResponse{}, err + } + m.attachDatabaseToChangelog(&response, route.DatabaseID) + return response, nil +} + +func (m *DatabaseManager) ListResolvedChangelog(ctx context.Context, workspace string, req ChangelogListRequest) (ChangelogListResponse, error) { + service, _, route, err := m.resolveWorkspace(ctx, workspace) + if err != nil { + return ChangelogListResponse{}, err + } + response, err := service.ListChangelog(ctx, route.WorkspaceID, req) + if err != nil { + return ChangelogListResponse{}, err + } + m.attachDatabaseToChangelog(&response, route.DatabaseID) + return response, nil } // GetSessionChangelogSummary returns the per-session rollup (op counts, delta bytes). @@ -1140,6 +1158,14 @@ func (m *DatabaseManager) GetSessionChangelogSummary(ctx context.Context, databa return service.GetSessionChangelogSummary(ctx, route.WorkspaceID, sessionID) } +func (m *DatabaseManager) GetResolvedSessionChangelogSummary(ctx context.Context, workspace, sessionID string) (SessionChangelogSummary, error) { + service, _, route, err := m.resolveWorkspace(ctx, workspace) + if err != nil { + return SessionChangelogSummary{}, err + } + return service.GetSessionChangelogSummary(ctx, route.WorkspaceID, sessionID) +} + // GetPathLastWriter returns the last-writer metadata for a single path. func (m *DatabaseManager) GetPathLastWriter(ctx context.Context, databaseID, workspace, path string) (PathLastWriter, error) { service, _, route, err := m.resolveScopedWorkspace(ctx, databaseID, workspace) @@ -1149,6 +1175,14 @@ func (m *DatabaseManager) GetPathLastWriter(ctx context.Context, databaseID, wor return service.GetPathLastWriter(ctx, route.WorkspaceID, path) } +func (m *DatabaseManager) GetResolvedPathLastWriter(ctx context.Context, workspace, path string) (PathLastWriter, error) { + service, _, route, err := m.resolveWorkspace(ctx, workspace) + if err != nil { + return PathLastWriter{}, err + } + return service.GetPathLastWriter(ctx, route.WorkspaceID, path) +} + func (m *DatabaseManager) ForkWorkspace(ctx context.Context, databaseID, sourceWorkspace, newWorkspace string) error { service, _, route, err := m.resolveScopedWorkspace(ctx, databaseID, sourceWorkspace) if err != nil { diff --git a/internal/controlplane/http.go b/internal/controlplane/http.go index e1b463f..0319e7a 100644 --- a/internal/controlplane/http.go +++ b/internal/controlplane/http.go @@ -1931,6 +1931,64 @@ func handleResolvedWorkspaceRoute( return } writeJSON(w, http.StatusOK, response) + case strings.HasSuffix(workspacePath, "/changes"): + workspace := strings.TrimSuffix(workspacePath, "/changes") + if r.Method != http.MethodGet { + writeError(w, fmt.Errorf("%s not allowed", r.Method)) + return + } + limit, err := queryInt(r, "limit", 100) + if err != nil { + writeError(w, err) + return + } + req := ChangelogListRequest{ + SessionID: strings.TrimSpace(r.URL.Query().Get("session_id")), + Path: strings.TrimSpace(r.URL.Query().Get("path")), + Since: strings.TrimSpace(r.URL.Query().Get("since")), + Until: strings.TrimSpace(r.URL.Query().Get("until")), + Limit: limit, + Reverse: strings.EqualFold(r.URL.Query().Get("direction"), "desc"), + } + response, err := manager.ListResolvedChangelog(r.Context(), workspace, req) + if err != nil { + writeError(w, err) + return + } + writeJSON(w, http.StatusOK, response) + case strings.Contains(workspacePath, "/sessions/") && strings.HasSuffix(workspacePath, "/summary"): + parts := strings.Split(strings.Trim(workspacePath, "/"), "/") + if len(parts) != 4 || parts[1] != "sessions" || parts[3] != "summary" { + writeError(w, os.ErrNotExist) + return + } + if r.Method != http.MethodGet { + writeError(w, fmt.Errorf("%s not allowed", r.Method)) + return + } + response, err := manager.GetResolvedSessionChangelogSummary(r.Context(), parts[0], parts[2]) + if err != nil { + writeError(w, err) + return + } + writeJSON(w, http.StatusOK, response) + case strings.HasSuffix(workspacePath, "/path-last"): + workspace := strings.TrimSuffix(workspacePath, "/path-last") + if r.Method != http.MethodGet { + writeError(w, fmt.Errorf("%s not allowed", r.Method)) + return + } + path := strings.TrimSpace(r.URL.Query().Get("path")) + if path == "" { + writeError(w, fmt.Errorf("path query parameter is required")) + return + } + response, err := manager.GetResolvedPathLastWriter(r.Context(), workspace, path) + if err != nil { + writeError(w, err) + return + } + writeJSON(w, http.StatusOK, response) case strings.HasSuffix(workspacePath, "/mcp-tokens"): workspace := strings.TrimSuffix(workspacePath, "/mcp-tokens") switch r.Method { diff --git a/internal/controlplane/http_test.go b/internal/controlplane/http_test.go index 9c2b104..021d8ff 100644 --- a/internal/controlplane/http_test.go +++ b/internal/controlplane/http_test.go @@ -714,6 +714,108 @@ func TestHTTPFileHistoryAndVersionContentRoutes(t *testing.T) { } } +func TestHTTPWorkspaceFirstChangeRoutesResolveOpaqueWorkspaceID(t *testing.T) { + t.Helper() + + ctx := context.Background() + manager, databaseID := newTestManager(t) + detail, err := manager.GetWorkspace(ctx, databaseID, "repo") + if err != nil { + t.Fatalf("GetWorkspace() returned error: %v", err) + } + if !strings.HasPrefix(detail.ID, "ws_") { + t.Fatalf("detail.ID = %q, want opaque workspace id", detail.ID) + } + + service, _, err := manager.serviceFor(ctx, databaseID) + if err != nil { + t.Fatalf("serviceFor() returned error: %v", err) + } + + const ( + sessionID = "sess-workspace-first" + path = "/notes/opaque-route.txt" + ) + WriteChangeEntries(ctx, service.store.rdb, detail.ID, []ChangeEntry{{ + SessionID: sessionID, + AgentID: "agt-http", + Op: ChangeOpPut, + Path: path, + DeltaBytes: 17, + ContentHash: "blob-opaque", + Source: ChangeSourceMCP, + }}) + + server := httptest.NewServer(NewHandler(manager, "*")) + defer server.Close() + + resp, err := http.Get(server.URL + "/v1/workspaces/" + detail.ID + "/changes?limit=10") + if err != nil { + t.Fatalf("GET workspace-first changes returned error: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + t.Fatalf("GET workspace-first changes status = %d, want %d, body=%s", resp.StatusCode, http.StatusOK, body) + } + + var changes ChangelogListResponse + if err := json.NewDecoder(resp.Body).Decode(&changes); err != nil { + t.Fatalf("Decode(workspace-first changes) returned error: %v", err) + } + if len(changes.Entries) != 1 { + t.Fatalf("len(workspace-first changes.entries) = %d, want 1", len(changes.Entries)) + } + if changes.Entries[0].Path != path || changes.Entries[0].SessionID != sessionID { + t.Fatalf("workspace-first change entry = %#v, want path/session %q/%q", changes.Entries[0], path, sessionID) + } + if changes.Entries[0].DatabaseID != databaseID { + t.Fatalf("workspace-first change database_id = %q, want %q", changes.Entries[0].DatabaseID, databaseID) + } + + resp, err = http.Get(server.URL + "/v1/workspaces/" + detail.ID + "/path-last?path=" + url.QueryEscape(path)) + if err != nil { + t.Fatalf("GET workspace-first path-last returned error: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + t.Fatalf("GET workspace-first path-last status = %d, want %d, body=%s", resp.StatusCode, http.StatusOK, body) + } + + var last PathLastWriter + if err := json.NewDecoder(resp.Body).Decode(&last); err != nil { + t.Fatalf("Decode(workspace-first path-last) returned error: %v", err) + } + if last.Path != path || last.SessionID != sessionID || last.Op != ChangeOpPut || last.ContentHash != "blob-opaque" { + t.Fatalf("workspace-first path-last = %#v, want path/session/op/hash %q/%q/%q/%q", last, path, sessionID, ChangeOpPut, "blob-opaque") + } + + resp, err = http.Get(server.URL + "/v1/workspaces/" + detail.ID + "/sessions/" + sessionID + "/summary") + if err != nil { + t.Fatalf("GET workspace-first session summary returned error: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + t.Fatalf("GET workspace-first session summary status = %d, want %d, body=%s", resp.StatusCode, http.StatusOK, body) + } + + var summary SessionChangelogSummary + if err := json.NewDecoder(resp.Body).Decode(&summary); err != nil { + t.Fatalf("Decode(workspace-first session summary) returned error: %v", err) + } + if summary.SessionID != sessionID { + t.Fatalf("workspace-first session summary session_id = %q, want %q", summary.SessionID, sessionID) + } + if summary.OpCounts[ChangeOpPut] != 1 { + t.Fatalf("workspace-first session summary put count = %d, want 1", summary.OpCounts[ChangeOpPut]) + } + if summary.DeltaBytes != 17 { + t.Fatalf("workspace-first session summary delta_bytes = %d, want 17", summary.DeltaBytes) + } +} + func TestHTTPUpdateWorkspaceRenamesOpaqueWorkspace(t *testing.T) { t.Helper() From 9d6d0df6024864229ca644aa1e95d7750f771e47 Mon Sep 17 00:00:00 2001 From: Andrew Brookins Date: Tue, 5 May 2026 23:31:08 -0700 Subject: [PATCH 3/3] Fix config-scoped state and unchanged checkpoints (#12) --- cmd/afs/afs_live_workspace.go | 6 +- cmd/afs/afs_mcp.go | 5 +- cmd/afs/afs_mcp_test.go | 37 ++++++++++++ cmd/afs/config_state.go | 39 +++++++++++-- cmd/afs/sync_integration_test.go | 71 ++++++++++++++++++++++++ internal/controlplane/mcp_hosted.go | 5 +- internal/controlplane/mcp_hosted_test.go | 37 ++++++++++++ sdk/python/tests/test_client.py | 17 +++++- sdk/typescript/test/sdk.test.mjs | 48 ++++++++++++++++ 9 files changed, 255 insertions(+), 10 deletions(-) diff --git a/cmd/afs/afs_live_workspace.go b/cmd/afs/afs_live_workspace.go index 6be4e72..88ddde6 100644 --- a/cmd/afs/afs_live_workspace.go +++ b/cmd/afs/afs_live_workspace.go @@ -120,9 +120,13 @@ func saveLiveWorkspaceCheckpoint(ctx context.Context, store *afsStore, workspace if err != nil { return false, err } + var metadata controlplane.SaveCheckpointFromLiveOptions + if len(options) > 0 { + metadata = options[0] + } if dirty, known, err := store.workspaceRootDirtyState(ctx, workspace); err != nil { return false, err - } else if known && !dirty { + } else if known && !dirty && !metadata.AllowUnchanged { if printResult { fmt.Println("No changes to save") } diff --git a/cmd/afs/afs_mcp.go b/cmd/afs/afs_mcp.go index b502d2f..457886a 100644 --- a/cmd/afs/afs_mcp.go +++ b/cmd/afs/afs_mcp.go @@ -826,8 +826,9 @@ func (s *afsMCPServer) toolCheckpointCreate(ctx context.Context, args map[string return nil, err } saved, err := saveAFSWorkspaceOrLiveRoot(ctx, s.cfg, s.store, workspace, checkpointID, false, controlplane.SaveCheckpointFromLiveOptions{ - Kind: controlplane.CheckpointKindManual, - Source: controlplane.CheckpointSourceMCP, + Kind: controlplane.CheckpointKindManual, + Source: controlplane.CheckpointSourceMCP, + AllowUnchanged: true, }) if err != nil { return nil, err diff --git a/cmd/afs/afs_mcp_test.go b/cmd/afs/afs_mcp_test.go index 0e76969..609f007 100644 --- a/cmd/afs/afs_mcp_test.go +++ b/cmd/afs/afs_mcp_test.go @@ -219,6 +219,43 @@ func TestAFSMCPCheckpointCreatePersistsPendingWrite(t *testing.T) { } } +func TestAFSMCPCheckpointCreateAllowsUnchangedWorkspace(t *testing.T) { + t.Helper() + + server, closeFn := setupAFSMCPTestServer(t) + defer closeFn() + server.profile = controlplane.MCPProfileWorkspaceRWCheckpoint + + checkpointResult := server.callTool(context.Background(), "checkpoint_create", map[string]any{ + "checkpoint": "unchanged-head", + }) + if checkpointResult.IsError { + t.Fatalf("checkpoint_create on unchanged workspace returned error result: %+v", checkpointResult) + } + + var checkpointPayload map[string]any + if err := decodeStructuredContent(checkpointResult.StructuredContent, &checkpointPayload); err != nil { + t.Fatalf("decodeStructuredContent(checkpoint unchanged) returned error: %v", err) + } + if created, _ := checkpointPayload["created"].(bool); !created { + t.Fatalf("checkpoint_create created = %#v, want true", checkpointPayload["created"]) + } + if checkpoint, _ := checkpointPayload["checkpoint"].(string); checkpoint != "unchanged-head" { + t.Fatalf("checkpoint_create checkpoint = %#v, want %q", checkpointPayload["checkpoint"], "unchanged-head") + } + + if _, err := server.store.getSavepointMeta(context.Background(), "repo", "unchanged-head"); err != nil { + t.Fatalf("getSavepointMeta(unchanged-head) returned error: %v", err) + } + + restoreResult := server.callTool(context.Background(), "checkpoint_restore", map[string]any{ + "checkpoint": "unchanged-head", + }) + if restoreResult.IsError { + t.Fatalf("checkpoint_restore after unchanged create returned error result: %+v", restoreResult) + } +} + func TestAFSMCPFileWriteDoesNotRematerializeLocalWorkspaceCache(t *testing.T) { t.Helper() diff --git a/cmd/afs/config_state.go b/cmd/afs/config_state.go index 5a5bd86..af74c9e 100644 --- a/cmd/afs/config_state.go +++ b/cmd/afs/config_state.go @@ -2,6 +2,7 @@ package main import ( "crypto/rand" + "crypto/sha256" "encoding/hex" "encoding/json" "errors" @@ -19,6 +20,10 @@ func configPath() string { if cfgPathOverride != "" { return cfgPathOverride } + return defaultConfigPath() +} + +func defaultConfigPath() string { exe, err := executablePath() if err != nil { return "afs.config.json" @@ -477,24 +482,50 @@ func defaultWorkRoot() string { return filepath.Join(stateDir(), "workspaces") } -func statePath() string { +func defaultStatePath() string { return filepath.Join(stateDir(), "state.json") } +func statePathForConfig(configFile string) string { + cleanConfig := cleanConfigPath(configFile) + if cleanConfig == "" || cleanConfig == cleanConfigPath(defaultConfigPath()) { + return defaultStatePath() + } + sum := sha256.Sum256([]byte(cleanConfig)) + return filepath.Join(stateDir(), "configs", hex.EncodeToString(sum[:8])+".json") +} + +func statePath() string { + return statePathForConfig(configPath()) +} + func saveState(st state) error { - if err := os.MkdirAll(stateDir(), 0o700); err != nil { + target := statePath() + if err := os.MkdirAll(filepath.Dir(target), 0o700); err != nil { return err } b, err := json.MarshalIndent(st, "", " ") if err != nil { return err } - return os.WriteFile(statePath(), b, 0o600) + return os.WriteFile(target, b, 0o600) } func loadState() (state, error) { + if st, err := loadStateFromPath(statePath()); err == nil { + return st, nil + } else if !errors.Is(err, os.ErrNotExist) { + return state{}, err + } + if sameConfigPath(statePath(), defaultStatePath()) { + return state{}, os.ErrNotExist + } + return loadStateFromPath(defaultStatePath()) +} + +func loadStateFromPath(path string) (state, error) { var st state - b, err := os.ReadFile(statePath()) + b, err := os.ReadFile(path) if err != nil { return st, err } diff --git a/cmd/afs/sync_integration_test.go b/cmd/afs/sync_integration_test.go index 8919c59..a218737 100644 --- a/cmd/afs/sync_integration_test.go +++ b/cmd/afs/sync_integration_test.go @@ -1194,6 +1194,77 @@ func TestCmdFileCreateExclusiveRoundTrip(t *testing.T) { } } +func TestCmdFileCreateExclusiveUsesConfigScopedState(t *testing.T) { + t.Helper() + + env := newSyncTestEnv(t) + env.startDaemon(t) + defer env.stopDaemon() + + oldCfgPathOverride := cfgPathOverride + cfgPathOverride = filepath.Join(t.TempDir(), "afs.config.json") + t.Cleanup(func() { + cfgPathOverride = oldCfgPathOverride + }) + + cfg := defaultConfig() + cfg.ProductMode = productModeLocal + cfg.Mode = modeSync + cfg.RedisAddr = env.mr.Addr() + cfg.RedisDB = 0 + cfg.LocalPath = env.localRoot + cfg.CurrentWorkspace = env.workspace + if err := saveConfig(cfg); err != nil { + t.Fatalf("saveConfig() returned error: %v", err) + } + + st := state{ + ProductMode: productModeLocal, + RedisAddr: env.mr.Addr(), + RedisDB: 0, + CurrentWorkspace: env.workspace, + LocalPath: env.localRoot, + Mode: modeSync, + SyncPID: os.Getpid(), + } + if err := saveState(st); err != nil { + t.Fatalf("saveState() returned error: %v", err) + } + if sameConfigPath(statePath(), defaultStatePath()) { + t.Fatalf("statePath() = %q, want config-scoped path distinct from legacy %q", statePath(), defaultStatePath()) + } + + legacyState := state{ + ProductMode: productModeLocal, + RedisAddr: "127.0.0.1:1", + RedisDB: 99, + CurrentWorkspace: "legacy-workspace", + LocalPath: t.TempDir(), + Mode: modeSync, + SyncPID: os.Getpid(), + } + rawLegacyState, err := json.MarshalIndent(legacyState, "", " ") + if err != nil { + t.Fatalf("json.MarshalIndent(legacyState) returned error: %v", err) + } + if err := os.MkdirAll(filepath.Dir(defaultStatePath()), 0o700); err != nil { + t.Fatalf("MkdirAll(defaultStatePath dir) returned error: %v", err) + } + if err := os.WriteFile(defaultStatePath(), rawLegacyState, 0o600); err != nil { + t.Fatalf("WriteFile(defaultStatePath) returned error: %v", err) + } + + if err := cmdFS([]string{"fs", "create-exclusive", "--content", "agent-c\n", "/tasks/003.claim"}); err != nil { + t.Fatalf("cmdFS(create-exclusive with config-scoped state) returned error: %v", err) + } + assertEventually(t, 3*time.Second, "remote 003.claim", func() bool { + return env.remoteExists(t, "tasks/003.claim") + }) + if got := env.readRemoteFile(t, "tasks/003.claim"); got != "agent-c\n" { + t.Fatalf("remote content = %q, want %q", got, "agent-c\n") + } +} + // Scenario 1 (burst variant): a batch of files written before startup all // land remotely, and the steady-state has no spurious echo loops. func TestSyncStartupUploadBurst(t *testing.T) { diff --git a/internal/controlplane/mcp_hosted.go b/internal/controlplane/mcp_hosted.go index 7a4c18c..377f2ce 100644 --- a/internal/controlplane/mcp_hosted.go +++ b/internal/controlplane/mcp_hosted.go @@ -549,8 +549,9 @@ func (p *hostedMCPProvider) callWorkspaceTool(ctx context.Context, name string, if err = validateHostedMCPName("checkpoint", checkpointID); err == nil { var saved bool saved, err = p.manager.SaveCheckpointFromLiveWithOptions(ctx, p.databaseID, p.workspace, checkpointID, SaveCheckpointFromLiveOptions{ - Kind: CheckpointKindManual, - Source: CheckpointSourceMCP, + Kind: CheckpointKindManual, + Source: CheckpointSourceMCP, + AllowUnchanged: true, }) value = map[string]any{ "workspace": p.workspace, diff --git a/internal/controlplane/mcp_hosted_test.go b/internal/controlplane/mcp_hosted_test.go index d13bc4c..81868dc 100644 --- a/internal/controlplane/mcp_hosted_test.go +++ b/internal/controlplane/mcp_hosted_test.go @@ -79,6 +79,43 @@ func TestHostedMCPFileCreateExclusiveFailsWhenFileExists(t *testing.T) { } } +func TestHostedMCPCheckpointCreateAllowsUnchangedWorkspace(t *testing.T) { + t.Helper() + + manager, databaseID := newTestManager(t) + provider := &hostedMCPProvider{ + manager: manager, + databaseID: databaseID, + workspace: "repo", + profile: MCPProfileWorkspaceRWCheckpoint, + } + + checkpointResult := provider.CallTool(context.Background(), "checkpoint_create", map[string]any{ + "checkpoint": "unchanged-head", + }) + if checkpointResult.IsError { + t.Fatalf("checkpoint_create on unchanged workspace returned error result: %+v", checkpointResult) + } + + var checkpointPayload map[string]any + if err := decodeHostedStructuredContent(checkpointResult.StructuredContent, &checkpointPayload); err != nil { + t.Fatalf("decodeHostedStructuredContent(checkpoint unchanged) returned error: %v", err) + } + if created, _ := checkpointPayload["created"].(bool); !created { + t.Fatalf("checkpoint_create created = %#v, want true", checkpointPayload["created"]) + } + if checkpoint, _ := checkpointPayload["checkpoint"].(string); checkpoint != "unchanged-head" { + t.Fatalf("checkpoint_create checkpoint = %#v, want %q", checkpointPayload["checkpoint"], "unchanged-head") + } + + restoreResult := provider.CallTool(context.Background(), "checkpoint_restore", map[string]any{ + "checkpoint": "unchanged-head", + }) + if restoreResult.IsError { + t.Fatalf("checkpoint_restore after unchanged create returned error result: %+v", restoreResult) + } +} + func TestHostedMCPWorkspaceVersioningPolicyToolsRoundTrip(t *testing.T) { t.Helper() diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index e7c1fef..3772767 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -1,7 +1,7 @@ import unittest from unittest.mock import patch -from redis_afs.client import AFSError, FSClient, MCPHttpClient, MountedFS, _MountedWorkspace, _normalize_mcp_endpoint +from redis_afs.client import AFSError, CheckpointClient, FSClient, MCPHttpClient, MountedFS, _MountedWorkspace, _normalize_mcp_endpoint class FakeMCP: @@ -32,6 +32,8 @@ def call_tool(self, name, arguments=None): return {"entries": entries} if name == "checkpoint_create": return {"workspace": "workspace", "checkpoint": arguments.get("checkpoint") or "auto", "created": True} + if name == "checkpoint_restore": + return {"workspace": "workspace", "checkpoint": arguments["checkpoint"], "restored": True} raise AssertionError(f"unexpected tool {name}") @@ -90,6 +92,8 @@ def call_tool(self, name, arguments=None): return {"entries": entries} if name == "checkpoint_create": return {"workspace": "workspace", "checkpoint": arguments.get("checkpoint") or "auto", "created": True} + if name == "checkpoint_restore": + return {"workspace": "workspace", "checkpoint": arguments["checkpoint"], "restored": True} raise AssertionError(f"unexpected tool {name}") @@ -144,6 +148,17 @@ def test_fs_mount_issues_workspace_token_and_reads_and_writes_files(self): class EndpointTest(unittest.TestCase): + def test_checkpoint_create_and_restore_round_trip(self): + checkpoint = CheckpointClient(FakeMCP()) + + created = checkpoint.create(workspace="repo", checkpoint="unchanged-head") + restored = checkpoint.restore(workspace="repo", checkpoint="unchanged-head") + + self.assertTrue(created["created"]) + self.assertEqual(created["checkpoint"], "unchanged-head") + self.assertTrue(restored["restored"]) + self.assertEqual(restored["checkpoint"], "unchanged-head") + def test_normalizes_mcp_endpoint(self): self.assertEqual(_normalize_mcp_endpoint("https://afs.cloud"), "https://afs.cloud/mcp") self.assertEqual(_normalize_mcp_endpoint("https://afs.cloud/mcp"), "https://afs.cloud/mcp") diff --git a/sdk/typescript/test/sdk.test.mjs b/sdk/typescript/test/sdk.test.mjs index a827877..f7e4b72 100644 --- a/sdk/typescript/test/sdk.test.mjs +++ b/sdk/typescript/test/sdk.test.mjs @@ -56,3 +56,51 @@ test("single-workspace mounts allow workspace-relative paths", async () => { assert.equal(await fs.readFile("/foobar/src/README.md"), "hello"); assert.deepEqual(fs.workspaceNames, ["foobar"]); }); + +test("checkpoint.create and checkpoint.restore round-trip through MCP", async () => { + const calls = []; + const afs = new AFS({ + apiKey: "test", + baseUrl: "https://afs.cloud", + fetch: async (_url, init) => { + const body = JSON.parse(String(init.body)); + calls.push(body); + let structuredContent; + if (body.params.name === "checkpoint_create") { + structuredContent = { + workspace: body.params.arguments.workspace, + checkpoint: body.params.arguments.checkpoint, + created: true, + }; + } else if (body.params.name === "checkpoint_restore") { + structuredContent = { + workspace: body.params.arguments.workspace, + checkpoint: body.params.arguments.checkpoint, + restored: true, + }; + } else { + throw new Error(`unexpected tool ${body.params.name}`); + } + return new Response( + JSON.stringify({ + jsonrpc: "2.0", + id: body.id, + result: { structuredContent }, + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + }, + }); + + const created = await afs.checkpoint.create({ workspace: "repo", checkpoint: "unchanged-head" }); + const restored = await afs.checkpoint.restore({ workspace: "repo", checkpoint: "unchanged-head" }); + + assert.equal(created.created, true); + assert.equal(created.checkpoint, "unchanged-head"); + assert.equal(restored.restored, true); + assert.equal(restored.checkpoint, "unchanged-head"); + assert.deepEqual( + calls.map((call) => call.params.name), + ["checkpoint_create", "checkpoint_restore"], + ); +});