diff --git a/apps/daemon/internal/agent/claudesdk/bridge_output.go b/apps/daemon/internal/agent/claudesdk/bridge_output.go index 6101240af..e80cd22fe 100644 --- a/apps/daemon/internal/agent/claudesdk/bridge_output.go +++ b/apps/daemon/internal/agent/claudesdk/bridge_output.go @@ -16,7 +16,7 @@ func (s *session) bridgeOutput() *bridgeOutput { scanner.Buffer(make([]byte, 64*1024), 2*1024*1024) for scanner.Scan() { raw := append([]byte(nil), scanner.Bytes()...) - if !s.receiveWorkspaceRead(raw) && !s.receiveWorkspaceDirectory(raw) { + if !s.receiveWorkspaceDirectory(raw) { output.frames <- raw } } diff --git a/apps/daemon/internal/agent/claudesdk/contracts.go b/apps/daemon/internal/agent/claudesdk/contracts.go index c432e9b34..c35ebb6a8 100644 --- a/apps/daemon/internal/agent/claudesdk/contracts.go +++ b/apps/daemon/internal/agent/claudesdk/contracts.go @@ -13,10 +13,8 @@ var ( _ agent.FunctionResultSubmitter = (*session)(nil) _ agent.PermissionResponder = (*session)(nil) _ agent.UserChoiceResponder = (*session)(nil) - _ agent.WorkspaceReader = (*session)(nil) _ agent.WorkspaceDirectoryLister = (*session)(nil) _ agent.WorkspaceWriter = (*session)(nil) - _ agent.WorkspaceReader = (*executor)(nil) _ agent.WorkspaceDirectoryLister = (*executor)(nil) _ agent.WorkspaceWriter = (*executor)(nil) ) diff --git a/apps/daemon/internal/agent/claudesdk/executor.go b/apps/daemon/internal/agent/claudesdk/executor.go index 664579186..31934e1f6 100644 --- a/apps/daemon/internal/agent/claudesdk/executor.go +++ b/apps/daemon/internal/agent/claudesdk/executor.go @@ -69,7 +69,6 @@ func startExecutor(ctx context.Context, checked *runtimeCheckCache, probe Config if err != nil { return nil, err } - base.reads.supported = slices.Contains(info.Features, "workspace_read") base.directories.supported = slices.Contains(info.Features, "workspace_directory") e := &executor{base: base, start: start, ready: make(chan error, 1), done: make(chan struct{}), nativeID: start.Resume} go e.read() @@ -180,7 +179,6 @@ func (e *executor) read() { } <-stderrDone _ = e.base.process.Wait() - e.base.stopWorkspaceReads() e.base.stopWorkspaceDirectories() e.mu.Lock() e.invalid = true @@ -320,9 +318,6 @@ func (s *session) invalidate() { } } -func (e *executor) ReadWorkspaceFile(ctx context.Context, path string, maxBytes int) (agent.WorkspaceReadResult, error) { - return e.base.ReadWorkspaceFile(ctx, path, maxBytes) -} func (e *executor) ListWorkspaceDirectory(ctx context.Context, path string, maxEntries int) (agent.WorkspaceDirectoryResult, error) { return e.base.ListWorkspaceDirectory(ctx, path, maxEntries) } diff --git a/apps/daemon/internal/agent/claudesdk/preparation_fixture_test.go b/apps/daemon/internal/agent/claudesdk/preparation_fixture_test.go index 081a199b9..b16f8ad93 100644 --- a/apps/daemon/internal/agent/claudesdk/preparation_fixture_test.go +++ b/apps/daemon/internal/agent/claudesdk/preparation_fixture_test.go @@ -38,7 +38,7 @@ func preparationRequest() proto.PromptRequestPayload { func runPreparationHelper() { mode := os.Getenv("SDK_HELPER_MODE") if len(os.Args) > 1 && strings.HasSuffix(os.Args[1], "runtime_check.js") { - features := []string{"workspace_tools", "workspace_prepare", "workspace_command_observations", "workspace_read", "workspace_directory"} + features := []string{"workspace_tools", "workspace_prepare", "workspace_command_observations", "workspace_directory"} if mode == "old-runtime" { features = []string{"workspace_tools"} } else if mode == "old-command-runtime" { @@ -89,10 +89,6 @@ func runPreparationHelper() { runWorkspaceDirectoryHelper(scanner, state, mode) return } - if strings.HasPrefix(mode, "read-") { - runWorkspaceReadHelper(scanner, state, mode) - return - } if !scanner.Scan() { return } diff --git a/apps/daemon/internal/agent/claudesdk/session.go b/apps/daemon/internal/agent/claudesdk/session.go index 0d8610264..770d212b8 100644 --- a/apps/daemon/internal/agent/claudesdk/session.go +++ b/apps/daemon/internal/agent/claudesdk/session.go @@ -23,7 +23,6 @@ type session struct { cancelOutput chan struct{} nativeEnded bool - reads workspaceReadState directories workspaceDirectoryState process *clirunner.Process writeMu *sync.Mutex diff --git a/apps/daemon/internal/agent/claudesdk/workspace_directory_live_linux_test.go b/apps/daemon/internal/agent/claudesdk/workspace_directory_live_linux_test.go index d6afad379..03445a78e 100644 --- a/apps/daemon/internal/agent/claudesdk/workspace_directory_live_linux_test.go +++ b/apps/daemon/internal/agent/claudesdk/workspace_directory_live_linux_test.go @@ -31,12 +31,8 @@ func liveWorkspaceDirectoryFixtures(t *testing.T, root string) { } } -func liveWorkspaceDirectories(t *testing.T, ctx context.Context, reader agent.WorkspaceReader, root, stage string) { +func liveWorkspaceDirectories(t *testing.T, ctx context.Context, lister agent.WorkspaceDirectoryLister, root, stage string) { t.Helper() - lister, ok := reader.(agent.WorkspaceDirectoryLister) - if !ok { - t.Fatal("workspace directory owner missing") - } result, err := lister.ListWorkspaceDirectory(ctx, "", 1000) if err != nil || result.Truncated { t.Fatal("workspace directory failed", stage, err) diff --git a/apps/daemon/internal/agent/claudesdk/workspace_live_linux_test.go b/apps/daemon/internal/agent/claudesdk/workspace_live_linux_test.go index d9a7e3321..c0c65d862 100644 --- a/apps/daemon/internal/agent/claudesdk/workspace_live_linux_test.go +++ b/apps/daemon/internal/agent/claudesdk/workspace_live_linux_test.go @@ -59,20 +59,19 @@ func TestLiveClaudeWorkspaceFactory(t *testing.T) { t.Fatal(err) } } - liveWorkspaceReadFixtures(t, config.Workspace.Directory) + liveWorkspaceDirectoryFixtures(t, config.Workspace.Directory) heartbeat := filepath.Join(config.Workspace.Directory, "heartbeat.txt") artifact := filepath.Join(config.Workspace.Directory, "value.txt") type evidence struct { - Reads []liveWorkspaceRead `json:"reads,omitempty"` - RunID string `json:"run_id"` - Events []proto.Envelope `json:"events"` - Done proto.DonePayload `json:"done"` - Failure string `json:"failure,omitempty"` - Cancelled bool `json:"cancelled"` - CancelMS int64 `json:"cancel_ms,omitempty"` - BridgePID int `json:"bridge_pid"` - Terminals int `json:"terminals"` - Heartbeats []string `json:"heartbeats,omitempty"` + RunID string `json:"run_id"` + Events []proto.Envelope `json:"events"` + Done proto.DonePayload `json:"done"` + Failure string `json:"failure,omitempty"` + Cancelled bool `json:"cancelled"` + CancelMS int64 `json:"cancel_ms,omitempty"` + BridgePID int `json:"bridge_pid"` + Terminals int `json:"terminals"` + Heartbeats []string `json:"heartbeats,omitempty"` } writeEvidence := func(name string, proof evidence) { t.Helper() @@ -106,7 +105,7 @@ func TestLiveClaudeWorkspaceFactory(t *testing.T) { s := running.(*session) defer s.Cancel(context.Background()) proof.BridgePID = s.process.Cmd.Process.Pid - proof.Reads = append(proof.Reads, liveWorkspaceReads(t, ctx, s, config.Workspace.Directory, "active", "read-binary.bin", "read-empty.bin", "read-large.bin")...) + liveWorkspaceDirectories(t, ctx, s, config.Workspace.Directory, "active") ticker := time.NewTicker(80 * time.Millisecond) defer ticker.Stop() for out != nil { @@ -116,7 +115,7 @@ func TestLiveClaudeWorkspaceFactory(t *testing.T) { case <-ticker.C: value, _ := os.ReadFile(heartbeat) if cancelOnEffect && !proof.Cancelled && len(value) > 0 && string(value) != "0" && string(value) != "1" { - proof.Reads = append(proof.Reads, liveWorkspaceReads(t, ctx, s, config.Workspace.Directory, "effect", "value.txt")...) + liveWorkspaceDirectories(t, ctx, s, config.Workspace.Directory, "effect") started := time.Now() if err := s.Cancel(ctx); err != nil { t.Fatal("factory cancellation failed", err) diff --git a/apps/daemon/internal/agent/claudesdk/workspace_read.go b/apps/daemon/internal/agent/claudesdk/workspace_read.go deleted file mode 100644 index ec5ea4a56..000000000 --- a/apps/daemon/internal/agent/claudesdk/workspace_read.go +++ /dev/null @@ -1,196 +0,0 @@ -package claudesdk - -import ( - "bytes" - "context" - "encoding/base64" - "encoding/json" - "io" - "strings" - "sync" - "time" - - "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" - "github.com/google/uuid" -) - -const workspaceReadMaxBytes = 1 << 20 -const workspaceReadTimeout = 12 * time.Second - -type workspaceReadState struct { - mu sync.Mutex - supported bool - closed bool - uncertain bool - pending *workspaceRead -} -type workspaceRead struct { - id string - maxBytes int - done chan struct{} - result agent.WorkspaceReadResult - err error -} -type workspaceReadEvent struct { - Type string `json:"type"` - ID string `json:"id"` - Data *string `json:"data_base64"` - Truncated *bool `json:"truncated"` - Error string `json:"error"` -} - -var _ agent.WorkspaceReader = (*session)(nil) - -func (s *session) ReadWorkspaceFile(ctx context.Context, path string, maxBytes int) (agent.WorkspaceReadResult, error) { - if s.owner != nil { - return s.owner.base.ReadWorkspaceFile(ctx, path, maxBytes) - } - read, err := s.admitWorkspaceRead(ctx, path, maxBytes) - if err != nil { - return agent.WorkspaceReadResult{}, err - } - return s.awaitWorkspaceRead(ctx, read) -} - -func (s *session) admitWorkspaceRead(ctx context.Context, path string, maxBytes int) (*workspaceRead, error) { - w := &s.reads - w.mu.Lock() - defer w.mu.Unlock() - if !w.supported { - return nil, agent.ErrWorkspaceReadUnsupported - } - if w.uncertain { - return nil, agent.ErrWorkspaceReadUncertain - } - if w.closed || ctx == nil || ctx.Err() != nil || s.process.Context().Err() != nil { - return nil, agent.ErrWorkspaceReadUnavailable - } - if maxBytes < 1 || maxBytes > workspaceReadMaxBytes || path == "" || len(path) > 8192 || strings.ContainsAny(path, "\x00\\\r\n") { - return nil, agent.ErrWorkspaceReadInvalid - } - for _, part := range strings.Split(path, "/") { - if part == "" || part == "." || part == ".." { - return nil, agent.ErrWorkspaceReadInvalid - } - } - if w.pending != nil { - return nil, agent.ErrWorkspaceReadBusy - } - read := &workspaceRead{id: uuid.NewString(), maxBytes: maxBytes, done: make(chan struct{})} - frame, err := json.Marshal(struct { - Type string `json:"type"` - ID string `json:"id"` - Path string `json:"path"` - MaxBytes int `json:"max_bytes"` - }{"workspace_read", read.id, path, maxBytes}) - if err != nil || len(frame)+1 > 8192 { - return nil, agent.ErrWorkspaceReadInvalid - } - w.pending = read - deadline := time.Now().Add(workspaceReadTimeout) - if requested, ok := ctx.Deadline(); ok && requested.Before(deadline) { - deadline = requested - } - go func() { - timer := time.NewTimer(time.Until(deadline)) - defer timer.Stop() - select { - case <-read.done: - return - case <-timer.C: - } - s.failWorkspaceRead(read) - }() - go func() { - s.writeMu.Lock() - _, err := s.process.Stdin.Write(append(frame, '\n')) - s.writeMu.Unlock() - if err != nil { - s.failWorkspaceRead(read) - } - }() - return read, nil -} - -func (s *session) failWorkspaceRead(read *workspaceRead) { - s.reads.mu.Lock() - pending := s.reads.pending == read - if pending { - s.reads.uncertain = true - s.reads.pending = nil - read.err = agent.ErrWorkspaceReadUncertain - s.process.Cancel() - close(read.done) - } - s.reads.mu.Unlock() -} - -func (s *session) awaitWorkspaceRead(_ context.Context, read *workspaceRead) (agent.WorkspaceReadResult, error) { - // Caller cancellation cannot discard an already admitted native wait. - <-read.done - return read.result, read.err -} - -func (s *session) receiveWorkspaceRead(raw []byte) bool { - var event workspaceReadEvent - if json.Unmarshal(raw, &event) != nil || event.Type != "workspace_read" { - return false - } - w := &s.reads - w.mu.Lock() - defer w.mu.Unlock() - read := w.pending - if read == nil || event.ID != read.id { - w.uncertain = true - s.process.Cancel() - return true - } - read.err = agent.ErrWorkspaceReadUncertain - decoder := json.NewDecoder(bytes.NewReader(raw)) - decoder.DisallowUnknownFields() - valid := decoder.Decode(&event) == nil && decoder.Decode(new(any)) == io.EOF - if !valid { - w.uncertain = true - w.pending = nil - close(read.done) - s.process.Cancel() - return true - } - if event.Error != "" && event.Data == nil && event.Truncated == nil { - switch event.Error { - case "invalid": - read.err = agent.ErrWorkspaceReadInvalid - case "busy": - read.err = agent.ErrWorkspaceReadBusy - case "unavailable": - read.err = agent.ErrWorkspaceReadUnavailable - } - } else if event.Error == "" && event.Data != nil && event.Truncated != nil { - data, err := base64.StdEncoding.Strict().DecodeString(*event.Data) - if err == nil && base64.StdEncoding.EncodeToString(data) == *event.Data && len(data) <= read.maxBytes && (!*event.Truncated || len(data) == read.maxBytes) { - read.result = agent.WorkspaceReadResult{Data: data, Truncated: *event.Truncated} - read.err = nil - } - } - if read.err == agent.ErrWorkspaceReadUncertain { - w.uncertain = true - s.process.Cancel() - } - w.pending = nil - close(read.done) - return true -} - -func (s *session) stopWorkspaceReads() { - w := &s.reads - w.mu.Lock() - defer w.mu.Unlock() - w.closed = true - if w.pending != nil { - read := w.pending - w.pending = nil - w.uncertain = true - read.err = agent.ErrWorkspaceReadUncertain - close(read.done) - } -} diff --git a/apps/daemon/internal/agent/claudesdk/workspace_read_live_linux_test.go b/apps/daemon/internal/agent/claudesdk/workspace_read_live_linux_test.go deleted file mode 100644 index e721126e7..000000000 --- a/apps/daemon/internal/agent/claudesdk/workspace_read_live_linux_test.go +++ /dev/null @@ -1,58 +0,0 @@ -//go:build linux - -package claudesdk - -import ( - "bytes" - "context" - "crypto/sha256" - "fmt" - "os" - "path/filepath" - "testing" - - "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" -) - -type liveWorkspaceRead struct { - Stage string `json:"stage"` - Path string `json:"path"` - Bytes int `json:"bytes"` - SHA256 string `json:"sha256"` - Truncated bool `json:"truncated"` -} - -func liveWorkspaceReadFixtures(t *testing.T, root string) { - t.Helper() - liveWorkspaceDirectoryFixtures(t, root) - for path, data := range map[string][]byte{"read-binary.bin": bytes.Repeat([]byte{0, 255, 128, 1}, 64), "read-empty.bin": {}, "read-large.bin": bytes.Repeat([]byte{0, 255, 128, 1}, (workspaceReadMaxBytes+40)/4)} { - if err := os.WriteFile(filepath.Join(root, path), data, 0600); err != nil { - t.Fatal(err) - } - } -} - -func liveWorkspaceReads(t *testing.T, ctx context.Context, reader agent.WorkspaceReader, root, stage string, paths ...string) []liveWorkspaceRead { - t.Helper() - liveWorkspaceDirectories(t, ctx, reader, root, stage) - var proof []liveWorkspaceRead - for _, path := range paths { - expected, err := os.ReadFile(filepath.Join(root, path)) - if err != nil { - t.Fatal(err) - } - result, err := reader.ReadWorkspaceFile(ctx, path, workspaceReadMaxBytes) - if err != nil { - t.Fatal("actual native workspace read failed", stage, path, err) - } - truncated := len(expected) > workspaceReadMaxBytes - if truncated { - expected = expected[:workspaceReadMaxBytes] - } - if !bytes.Equal(result.Data, expected) || result.Truncated != truncated { - t.Fatal("native bytes differ", stage, path, len(result.Data), result.Truncated) - } - proof = append(proof, liveWorkspaceRead{Stage: stage, Path: path, Bytes: len(result.Data), SHA256: fmt.Sprintf("%x", sha256.Sum256(result.Data)), Truncated: result.Truncated}) - } - return proof -} diff --git a/apps/daemon/internal/agent/claudesdk/workspace_read_test.go b/apps/daemon/internal/agent/claudesdk/workspace_read_test.go deleted file mode 100644 index 72d32a943..000000000 --- a/apps/daemon/internal/agent/claudesdk/workspace_read_test.go +++ /dev/null @@ -1,208 +0,0 @@ -//go:build unix - -package claudesdk - -import ( - "bufio" - "bytes" - "context" - "encoding/base64" - "encoding/json" - "errors" - "os" - "path/filepath" - "strings" - "testing" - "time" - - "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" - "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" -) - -func runWorkspaceReadHelper(scanner *bufio.Scanner, state, mode string) { - turnID := "" - for scanner.Scan() { - var req struct { - Type, ID, Path string - TurnID string `json:"turn_id"` - Max int `json:"max_bytes"` - } - _ = json.Unmarshal(scanner.Bytes(), &req) - if req.Type == "turn_cancel" { - reusable := false - _ = json.NewEncoder(os.Stdout).Encode(bridgeEvent{Type: "error", TurnID: turnID, Code: "cancelled"}) - confirmed := true - _ = json.NewEncoder(os.Stdout).Encode(bridgeEvent{Type: "turn_settled", Confirmed: &confirmed, TurnID: turnID, Reusable: &reusable, Reason: "fixture_closed"}) - return - } - if req.Type == "turn_start" { - turnID = req.TurnID - _ = json.NewEncoder(os.Stdout).Encode(bridgeEvent{Type: "turn_started", TurnID: turnID}) - _ = os.WriteFile(filepath.Join(state, "read-started"), []byte("{}"), 0600) - continue - } - _ = os.WriteFile(filepath.Join(state, "read-admitted"), []byte("{}"), 0600) - if mode == "read-held" { - for { - if _, err := os.Stat(filepath.Join(state, "read-release")); err == nil { - break - } - time.Sleep(time.Millisecond) - } - } - if mode == "read-exit" { - return - } - data := bytes.Repeat([]byte{0, 255, 1, 128}, req.Max/4) - if req.Path == "empty" { - data = nil - } - encoded := base64.StdEncoding.EncodeToString(data) - truncated := req.Path == "prefix" - response := map[string]any{"type": "workspace_read", "id": req.ID, "data_base64": encoded, "truncated": truncated} - switch req.Path { - case "uncertain": - response = map[string]any{"type": "workspace_read", "id": req.ID, "error": "uncertain"} - case "bad-base64": - response["data_base64"] = "!" - case "wrong-id": - response["id"] = "other" - case "extra": - response["extra"] = true - case "oversize": - response["data_base64"] = base64.StdEncoding.EncodeToString(make([]byte, req.Max+1)) - case "invalid": - response = map[string]any{"type": "workspace_read", "id": req.ID, "error": "invalid"} - } - _ = json.NewEncoder(os.Stdout).Encode(response) - } -} - -func readExecutor(t *testing.T, mode string) (*executor, Config) { - t.Helper() - config := preparationFixture(t, mode) - resource, err := NewExecutorFactory(config)(t.Context(), preparationRequest()) - if err != nil { - t.Fatal(err) - } - e := resource.(*executor) - t.Cleanup(func() { _ = e.Close(context.Background()) }) - return e, config -} - -func TestWorkspaceReadBoundsAndBinary(t *testing.T) { - e, _ := readExecutor(t, "read-normal") - for _, path := range []string{"", "/absolute", "../escape", "a//b", "a/./b", "a\\b", "a\x00b", strings.Repeat("界", 3000)} { - if _, err := e.ReadWorkspaceFile(t.Context(), path, 4); !errors.Is(err, agent.ErrWorkspaceReadInvalid) { - t.Fatalf("accepted %q: %v", path, err) - } - } - for _, limit := range []int{0, -1, workspaceReadMaxBytes + 1} { - if _, err := e.ReadWorkspaceFile(t.Context(), "file", limit); err != agent.ErrWorkspaceReadInvalid { - t.Fatal(limit, err) - } - } - for _, path := range []string{"file", "prefix", "empty"} { - result, err := e.ReadWorkspaceFile(t.Context(), path, workspaceReadMaxBytes) - expected := bytes.Repeat([]byte{0, 255, 1, 128}, workspaceReadMaxBytes/4) - if path == "empty" { - expected = nil - } - if err != nil || !bytes.Equal(result.Data, expected) || result.Truncated != (path == "prefix") { - t.Fatal(path, err, len(result.Data), result.Truncated) - } - } - if _, err := e.ReadWorkspaceFile(t.Context(), "invalid", 4); err != agent.ErrWorkspaceReadInvalid { - t.Fatal(err) - } - if _, err := e.ReadWorkspaceFile(t.Context(), "file", 4); err != nil { - t.Fatal(err) - } -} - -func TestWorkspaceReadDetachAndTurnStart(t *testing.T) { - e, config := readExecutor(t, "read-held") - ctx, detach := context.WithCancel(t.Context()) - defer detach() - result := make(chan error, 1) - go func() { _, err := e.ReadWorkspaceFile(ctx, "file", 4); result <- err }() - waitPreparationFile(t, filepath.Join(config.StateDir, "read-admitted")) - detach() - if _, err := e.ReadWorkspaceFile(t.Context(), "file", 4); err != agent.ErrWorkspaceReadBusy { - t.Fatal(err) - } - out := make(chan proto.Envelope, 16) - running, err := e.StartTurn(t.Context(), "run", proto.TextInput("hello"), out) - if err != nil { - t.Fatal(err) - } - select { - case err := <-result: - t.Fatal("caller detach discarded native wait", err) - default: - } - if err := os.WriteFile(filepath.Join(config.StateDir, "read-release"), nil, 0600); err != nil { - t.Fatal(err) - } - if err := <-result; err != nil { - t.Fatal(err) - } - waitPreparationFile(t, filepath.Join(config.StateDir, "read-started")) - if _, err := running.(agent.WorkspaceReader).ReadWorkspaceFile(t.Context(), "file", 4); err != nil { - t.Fatal(err) - } -} - -func TestWorkspaceReadUnknownAndRelease(t *testing.T) { - for _, path := range []string{"uncertain", "wrong-id", "bad-base64", "oversize", "extra"} { - t.Run(path, func(t *testing.T) { - e, _ := readExecutor(t, "read-normal") - result, err := e.ReadWorkspaceFile(t.Context(), path, 4) - if err != agent.ErrWorkspaceReadUncertain || len(result.Data) != 0 { - t.Fatal(result, err) - } - if _, err := e.ReadWorkspaceFile(t.Context(), "file", 4); err != agent.ErrWorkspaceReadUncertain && err != agent.ErrWorkspaceReadUnavailable { - t.Fatal(err) - } - }) - } - for _, mode := range []string{"read-held", "read-exit"} { - t.Run(mode, func(t *testing.T) { - e, config := readExecutor(t, mode) - done := make(chan error, 1) - go func() { _, err := e.ReadWorkspaceFile(t.Context(), "file", 4); done <- err }() - waitPreparationFile(t, filepath.Join(config.StateDir, "read-admitted")) - if mode == "read-held" { - if err := e.Close(t.Context()); err != nil { - t.Fatal(err) - } - } - select { - case err := <-done: - if err != agent.ErrWorkspaceReadUncertain { - t.Fatal(err) - } - case <-time.After(5 * time.Second): - t.Fatal("native waiter lost on release") - } - }) - } -} - -func TestWorkspaceReadDeadlineStopsOwnerBeforeUnknown(t *testing.T) { - e, config := readExecutor(t, "read-held") - ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) - defer cancel() - done := make(chan error, 1) - go func() { _, err := e.ReadWorkspaceFile(ctx, "file", 4); done <- err }() - waitPreparationFile(t, filepath.Join(config.StateDir, "read-admitted")) - if err := <-done; err != agent.ErrWorkspaceReadUncertain { - t.Fatal(err) - } - if e.base.process.Context().Err() == nil { - t.Fatal("uncertain deadline returned before owner cancellation") - } - if _, err := e.StartTurn(t.Context(), "late", proto.TextInput("hello"), make(chan proto.Envelope, 8)); err == nil { - t.Fatal("unknown owner accepted a new Turn") - } -} diff --git a/apps/daemon/internal/agent/codex/contracts.go b/apps/daemon/internal/agent/codex/contracts.go index 329b053c5..7a19d9eb8 100644 --- a/apps/daemon/internal/agent/codex/contracts.go +++ b/apps/daemon/internal/agent/codex/contracts.go @@ -13,10 +13,8 @@ var ( _ agent.FunctionResultSubmitter = (*Session)(nil) _ agent.PermissionResponder = (*Session)(nil) _ agent.UserChoiceResponder = (*Session)(nil) - _ agent.WorkspaceReader = (*Session)(nil) _ agent.WorkspaceDirectoryLister = (*Session)(nil) _ agent.WorkspaceWriter = (*Session)(nil) - _ agent.WorkspaceReader = (*Executor)(nil) _ agent.WorkspaceDirectoryLister = (*Executor)(nil) _ agent.WorkspaceWriter = (*Executor)(nil) ) diff --git a/apps/daemon/internal/agent/codex/unsupported.go b/apps/daemon/internal/agent/codex/unsupported.go index 3da26fb7f..3688145dc 100644 --- a/apps/daemon/internal/agent/codex/unsupported.go +++ b/apps/daemon/internal/agent/codex/unsupported.go @@ -5,10 +5,6 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" ) -func (s *Executor) ReadWorkspaceFile(context.Context, string, int) (agent.WorkspaceReadResult, error) { - return agent.WorkspaceReadResult{}, agent.ErrWorkspaceReadUnsupported -} - func (s *Executor) ListWorkspaceDirectory(context.Context, string, int) (agent.WorkspaceDirectoryResult, error) { return agent.WorkspaceDirectoryResult{}, agent.ErrWorkspaceReadUnsupported } @@ -17,10 +13,6 @@ func (s *Executor) WriteWorkspaceFile(context.Context, string, []byte) (agent.Wo return agent.WorkspaceWriteResult{}, agent.ErrWorkspaceWriteUnsupported } -func (s *Session) ReadWorkspaceFile(context.Context, string, int) (agent.WorkspaceReadResult, error) { - return agent.WorkspaceReadResult{}, agent.ErrWorkspaceReadUnsupported -} - func (s *Session) ListWorkspaceDirectory(context.Context, string, int) (agent.WorkspaceDirectoryResult, error) { return agent.WorkspaceDirectoryResult{}, agent.ErrWorkspaceReadUnsupported } diff --git a/apps/daemon/internal/agent/codex/unsupported_test.go b/apps/daemon/internal/agent/codex/unsupported_test.go index fe2fe3616..8a4234e13 100644 --- a/apps/daemon/internal/agent/codex/unsupported_test.go +++ b/apps/daemon/internal/agent/codex/unsupported_test.go @@ -24,13 +24,6 @@ func TestUnsupportedExtensionsHaveNoNativeEffects(t *testing.T) { t.Fatal("unsupported error disclosed input") } } - for _, owner := range []agent.WorkspaceReader{(*Executor)(nil), (*Session)(nil)} { - result, err := owner.ReadWorkspaceFile(ctx, secret, 1) - check(err) - if len(result.Data) != 0 || result.Truncated { - t.Fatal("unsupported read fabricated data") - } - } for _, owner := range []agent.WorkspaceDirectoryLister{(*Executor)(nil), (*Session)(nil)} { result, err := owner.ListWorkspaceDirectory(ctx, secret, 1) check(err) diff --git a/apps/daemon/internal/agent/contract_declarations_test.go b/apps/daemon/internal/agent/contract_declarations_test.go index 316f61634..a07c4e29e 100644 --- a/apps/daemon/internal/agent/contract_declarations_test.go +++ b/apps/daemon/internal/agent/contract_declarations_test.go @@ -24,7 +24,6 @@ func TestPublicHarnessContractDeclarations(t *testing.T) { "FunctionResultSubmitter": {"session"}, "PermissionResponder": {"session"}, "UserChoiceResponder": {"session"}, - "WorkspaceReader": {"executor", "session"}, "WorkspaceDirectoryLister": {"executor", "session"}, "WorkspaceWriter": {"executor", "session"}, } diff --git a/apps/daemon/internal/agent/harness.go b/apps/daemon/internal/agent/harness.go index cb7232fef..4f6bc8b26 100644 --- a/apps/daemon/internal/agent/harness.go +++ b/apps/daemon/internal/agent/harness.go @@ -621,11 +621,6 @@ type UserChoiceResponder interface { // A resource without native access returns the corresponding Unsupported error; // this does not disable capabilities provided by the common workspace owner. -// WorkspaceReader returns success only after acknowledged native close on an existing owner. -type WorkspaceReader interface { - ReadWorkspaceFile(context.Context, string, int) (WorkspaceReadResult, error) -} - // WorkspaceDirectoryLister reads one directory through an existing workspace owner. // An empty directory selects the root; other paths contain only relative components. // Entry names are single components. Kind is file, directory, symlink or other; diff --git a/apps/daemon/internal/agent/mcode/contracts.go b/apps/daemon/internal/agent/mcode/contracts.go index 03261a304..3df9bbe3d 100644 --- a/apps/daemon/internal/agent/mcode/contracts.go +++ b/apps/daemon/internal/agent/mcode/contracts.go @@ -13,10 +13,8 @@ var ( _ agent.FunctionResultSubmitter = (*Session)(nil) _ agent.PermissionResponder = (*Session)(nil) _ agent.UserChoiceResponder = (*Session)(nil) - _ agent.WorkspaceReader = (*Session)(nil) _ agent.WorkspaceDirectoryLister = (*Session)(nil) _ agent.WorkspaceWriter = (*Session)(nil) - _ agent.WorkspaceReader = (*executor)(nil) _ agent.WorkspaceDirectoryLister = (*executor)(nil) _ agent.WorkspaceWriter = (*executor)(nil) ) diff --git a/apps/daemon/internal/agent/mcode/unsupported.go b/apps/daemon/internal/agent/mcode/unsupported.go index e221bd595..2ee04cf90 100644 --- a/apps/daemon/internal/agent/mcode/unsupported.go +++ b/apps/daemon/internal/agent/mcode/unsupported.go @@ -11,10 +11,6 @@ func (s *Session) SubmitFunctionResult(context.Context, proto.FunctionResultPayl return fmt.Errorf("%w: native public function tools are not qualified", agent.ErrUnsupportedOperation) } -func (s *executor) ReadWorkspaceFile(context.Context, string, int) (agent.WorkspaceReadResult, error) { - return agent.WorkspaceReadResult{}, agent.ErrWorkspaceReadUnsupported -} - func (s *executor) ListWorkspaceDirectory(context.Context, string, int) (agent.WorkspaceDirectoryResult, error) { return agent.WorkspaceDirectoryResult{}, agent.ErrWorkspaceReadUnsupported } @@ -23,10 +19,6 @@ func (s *executor) WriteWorkspaceFile(context.Context, string, []byte) (agent.Wo return agent.WorkspaceWriteResult{}, agent.ErrWorkspaceWriteUnsupported } -func (s *Session) ReadWorkspaceFile(context.Context, string, int) (agent.WorkspaceReadResult, error) { - return agent.WorkspaceReadResult{}, agent.ErrWorkspaceReadUnsupported -} - func (s *Session) ListWorkspaceDirectory(context.Context, string, int) (agent.WorkspaceDirectoryResult, error) { return agent.WorkspaceDirectoryResult{}, agent.ErrWorkspaceReadUnsupported } diff --git a/apps/daemon/internal/agent/mcode/unsupported_test.go b/apps/daemon/internal/agent/mcode/unsupported_test.go index eb0609bb6..3e23d3c24 100644 --- a/apps/daemon/internal/agent/mcode/unsupported_test.go +++ b/apps/daemon/internal/agent/mcode/unsupported_test.go @@ -27,13 +27,6 @@ func TestUnsupportedExtensionsHaveNoNativeEffects(t *testing.T) { } var turn *Session check(turn.SubmitFunctionResult(ctx, proto.FunctionResultPayload{CallID: secret, DeliveryID: secret})) - for _, owner := range []agent.WorkspaceReader{(*executor)(nil), (*Session)(nil)} { - result, err := owner.ReadWorkspaceFile(ctx, secret, 1) - check(err) - if len(result.Data) != 0 || result.Truncated { - t.Fatal("unsupported read fabricated data") - } - } for _, owner := range []agent.WorkspaceDirectoryLister{(*executor)(nil), (*Session)(nil)} { result, err := owner.ListWorkspaceDirectory(ctx, secret, 1) check(err) diff --git a/apps/daemon/internal/agent/workspace_read.go b/apps/daemon/internal/agent/workspace_read.go index 71f1db3e5..7d4ebbe3b 100644 --- a/apps/daemon/internal/agent/workspace_read.go +++ b/apps/daemon/internal/agent/workspace_read.go @@ -5,11 +5,6 @@ import ( "fmt" ) -type WorkspaceReadResult struct { - Data []byte - Truncated bool -} - var ( ErrWorkspaceReadUnsupported = fmt.Errorf("%w: workspace read", ErrUnsupportedOperation) ErrWorkspaceReadUnavailable = errors.New("workspace read unavailable") diff --git a/apps/daemon/internal/dispatch/local_directory_test.go b/apps/daemon/internal/dispatch/local_directory_test.go index 03b2f69c2..e354b668b 100644 --- a/apps/daemon/internal/dispatch/local_directory_test.go +++ b/apps/daemon/internal/dispatch/local_directory_test.go @@ -48,19 +48,11 @@ func TestLocalDirectoryPreparationNeedsNoHarnessAndRejectsOtherOwners(t *testing t.Fatal(err) } ready := waitPreparationStatus(t, sender, "idle", "ready", "") - read := proto.WorkspaceReadPayload{EnvironmentID: environment, Handle: ready.Handle, Operation: "directory", MaxEntries: 10} + read := proto.WorkspaceReadPayload{EnvironmentID: environment, Handle: ready.Handle, MaxEntries: 10} _ = r.Handle(t.Context(), mustEnv(t, proto.TypeWorkspaceRead, "list", read)) if got := waitWorkspaceRead(t, sender, "list"); got.Outcome != "completed" || got.Directory == nil { t.Fatal("idle directory unavailable", got) } - if err := os.WriteFile(filepath.Join(workspace, "bytes"), []byte{0, 255, 17}, 0600); err != nil { - t.Fatal(err) - } - content := proto.WorkspaceReadPayload{EnvironmentID: environment, Handle: ready.Handle, Path: "bytes", MaxBytes: 2} - _ = r.Handle(t.Context(), mustEnv(t, proto.TypeWorkspaceRead, "content", content)) - if got := waitWorkspaceRead(t, sender, "content"); got.Outcome != "completed" || !got.Truncated || !got.CloseAcknowledged || len(got.Data) != 2 || got.Data[1] != 255 { - t.Fatal("file prefix unavailable", got) - } read.EnvironmentID = uuid.NewString() _ = r.Handle(t.Context(), mustEnv(t, proto.TypeWorkspaceRead, "foreign", read)) if got := waitWorkspaceRead(t, sender, "foreign"); got.Outcome != "rejected" { @@ -134,7 +126,7 @@ func TestLocalDirectoryKeepsNotDirectorySeparateFromFailures(t *testing.T) { "file": {Outcome: "rejected", ErrorCode: proto.WorkspaceReadNotDirectory}, "../invalid": {Outcome: "rejected", ErrorCode: "invalid_request"}, } { - read := proto.WorkspaceReadPayload{EnvironmentID: environment, Handle: ready.Handle, Operation: "directory", Path: path, MaxEntries: 10} + read := proto.WorkspaceReadPayload{EnvironmentID: environment, Handle: ready.Handle, Path: path, MaxEntries: 10} _ = r.Handle(t.Context(), mustEnv(t, proto.TypeWorkspaceRead, path, read)) if got := waitWorkspaceRead(t, sender, path); got.Outcome != want.Outcome || got.ErrorCode != want.ErrorCode || got.Directory != nil || got.CloseAcknowledged { t.Fatal("native directory result changed", path, got) diff --git a/apps/daemon/internal/dispatch/native_file_results_test.go b/apps/daemon/internal/dispatch/native_file_results_test.go index 2cd90fe56..3c9ca7d3d 100644 --- a/apps/daemon/internal/dispatch/native_file_results_test.go +++ b/apps/daemon/internal/dispatch/native_file_results_test.go @@ -21,7 +21,7 @@ func TestNativeDirectoryFailureMapping(t *testing.T) { {fs.ErrPermission, "rejected", "permission_denied"}, {agent.ErrWorkspaceReadUncertain, "unknown", "read_unconfirmed"}, } { - got := workspaceReadResult(agent.WorkspaceReadResult{}, test.err, 0) + got := workspaceReadFailure(test.err) if got.Outcome != test.outcome || got.ErrorCode != test.code { t.Fatal(test.err, got) } diff --git a/apps/daemon/internal/dispatch/preparation_executor_fixture_test.go b/apps/daemon/internal/dispatch/preparation_executor_fixture_test.go index 8ca7eebdc..a5966ce69 100644 --- a/apps/daemon/internal/dispatch/preparation_executor_fixture_test.go +++ b/apps/daemon/internal/dispatch/preparation_executor_fixture_test.go @@ -134,9 +134,3 @@ func (t *preparationTurn) SubmitPromptForUserChoice(ctx context.Context, id stri } return agent.ErrUnknownAsk } -func (t *preparationTurn) ReadWorkspaceFile(ctx context.Context, path string, limit int) (agent.WorkspaceReadResult, error) { - if target, ok := t.Session.(agent.WorkspaceReader); ok { - return target.ReadWorkspaceFile(ctx, path, limit) - } - return agent.WorkspaceReadResult{}, agent.ErrWorkspaceReadUnsupported -} diff --git a/apps/daemon/internal/dispatch/prepared_handoff_mutation_test.go b/apps/daemon/internal/dispatch/prepared_handoff_mutation_test.go index c2292f83c..7689961ed 100644 --- a/apps/daemon/internal/dispatch/prepared_handoff_mutation_test.go +++ b/apps/daemon/internal/dispatch/prepared_handoff_mutation_test.go @@ -17,7 +17,6 @@ type preparedMutationSession struct { cancelOnce sync.Once functions atomic.Int32 steers atomic.Int32 - reads atomic.Int32 } func (s *preparedMutationSession) Cancel(ctx context.Context) error { @@ -35,11 +34,6 @@ func (s *preparedMutationSession) Steer(context.Context, proto.PromptSteerPayloa return nil } -func (s *preparedMutationSession) ReadWorkspaceFile(context.Context, string, int) (agent.WorkspaceReadResult, error) { - s.reads.Add(1) - return agent.WorkspaceReadResult{Data: []byte("x")}, nil -} - type blockingPreparedReceiptSender struct { *recSender deliveryID string diff --git a/apps/daemon/internal/dispatch/prepared_handoff_test.go b/apps/daemon/internal/dispatch/prepared_handoff_test.go index b27b899ca..c4cf45c91 100644 --- a/apps/daemon/internal/dispatch/prepared_handoff_test.go +++ b/apps/daemon/internal/dispatch/prepared_handoff_test.go @@ -193,7 +193,7 @@ func TestPreparedHandoffDuplicateStartDoesNotReexecuteDuringPublication(t *testi if ack := lastSteeringAck(t, sender.recSender, "run", "publication-steering"); ack.ErrorCode != "not_ready" { t.Fatalf("pre-publication steering = %+v", ack) } - read := proto.WorkspaceReadPayload{RunID: "run", EnvironmentID: preparationEnvironmentID, Path: "file", MaxBytes: 1} + read := proto.WorkspaceReadPayload{RunID: "run", EnvironmentID: preparationEnvironmentID, MaxEntries: 1} if err := r.Handle(t.Context(), mustEnv(t, proto.TypeWorkspaceRead, "publication-read", read)); err != nil { t.Fatal(err) } @@ -205,7 +205,7 @@ func TestPreparedHandoffDuplicateStartDoesNotReexecuteDuringPublication(t *testi session.askMu.Lock() askCalls := len(session.askCalls) session.askMu.Unlock() - if session.functions.Load() != 0 || session.steers.Load() != 0 || session.reads.Load() != 0 || len(session.submissions()) != 0 || askCalls != 0 { + if session.functions.Load() != 0 || session.steers.Load() != 0 || len(session.submissions()) != 0 || askCalls != 0 { t.Fatal("private Session accepted work before started publication") } duplicate := mustEnv(t, proto.TypeExecutionStart, "request", proto.ExecutionStartPayload{Handle: ready.Handle, ExecutorID: ready.ExecutorID, RunID: "run", Input: proto.TextInput("input")}) diff --git a/apps/daemon/internal/dispatch/workspace_directory_test.go b/apps/daemon/internal/dispatch/workspace_directory_test.go index 49686cca6..bf09335c5 100644 --- a/apps/daemon/internal/dispatch/workspace_directory_test.go +++ b/apps/daemon/internal/dispatch/workspace_directory_test.go @@ -26,11 +26,11 @@ func TestWorkspaceDirectoryRetainsEnvironmentAndTransferredOwner(t *testing.T) { } _ = r.Handle(t.Context(), mustEnv(t, proto.TypeExecutionPrepare, "prepare", preparationRequest())) ready := waitPreparationStatus(t, sender, "prepare", "ready", "") - request := proto.WorkspaceReadPayload{Operation: "directory", Handle: ready.Handle, EnvironmentID: preparationEnvironmentID, MaxEntries: 1} + request := proto.WorkspaceReadPayload{Handle: ready.Handle, EnvironmentID: preparationEnvironmentID, MaxEntries: 1} for _, phase := range []string{"idle", "active"} { _ = r.Handle(t.Context(), mustEnv(t, proto.TypeWorkspaceRead, phase, request)) result := waitWorkspaceRead(t, sender, phase) - if result.Outcome != "completed" || !result.CloseAcknowledged || result.Directory == nil || !result.Directory.Truncated || len(result.Directory.Entries) != 1 || len(result.Data) != 0 { + if result.Outcome != "completed" || !result.CloseAcknowledged || result.Directory == nil || !result.Directory.Truncated || len(result.Directory.Entries) != 1 { t.Fatal(result) } bad := request @@ -50,10 +50,8 @@ func TestWorkspaceDirectoryRetainsEnvironmentAndTransferredOwner(t *testing.T) { } } for index, bad := range []proto.WorkspaceReadPayload{ - {Operation: "directory", RunID: "run", EnvironmentID: preparationEnvironmentID, MaxEntries: 2, MaxBytes: 1}, - {Operation: "directory", RunID: "run", EnvironmentID: preparationEnvironmentID, MaxEntries: proto.WorkspaceDirectoryMaxEntries + 1}, - {Operation: "directory", Handle: ready.Handle, RunID: "run", EnvironmentID: preparationEnvironmentID, MaxEntries: 2}, - {Operation: "recursive", RunID: "run", EnvironmentID: preparationEnvironmentID, MaxEntries: 2}, + {RunID: "run", EnvironmentID: preparationEnvironmentID, MaxEntries: proto.WorkspaceDirectoryMaxEntries + 1}, + {Handle: ready.Handle, RunID: "run", EnvironmentID: preparationEnvironmentID, MaxEntries: 2}, } { id := fmt.Sprintf("invalid-%d", index) _ = r.Handle(t.Context(), mustEnv(t, proto.TypeWorkspaceRead, id, bad)) diff --git a/apps/daemon/internal/dispatch/workspace_read.go b/apps/daemon/internal/dispatch/workspace_read.go index 7e4874bb8..4ef393ae4 100644 --- a/apps/daemon/internal/dispatch/workspace_read.go +++ b/apps/daemon/internal/dispatch/workspace_read.go @@ -36,7 +36,7 @@ func (r *Router) handleWorkspaceRead(ctx context.Context, env proto.Envelope) er r.mu.Unlock() return r.sendWorkspaceRead(ctx, env, rejectedWorkspaceRead("read_capacity")) } - resource, code := r.workspaceResourceLocked(env.Assignment, request) + lister, code := r.workspaceResourceLocked(env.Assignment, request) if code != "" { r.mu.Unlock() return r.sendWorkspaceRead(ctx, env, rejectedWorkspaceRead(code)) @@ -55,81 +55,66 @@ func (r *Router) handleWorkspaceRead(ctx context.Context, env proto.Envelope) er // Observer loss does not discard an admitted native wait or replay it. operation, cancel := context.WithTimeout(context.WithoutCancel(ctx), 12*time.Second) defer cancel() - result := executeWorkspaceRead(operation, resource, request) + result := listWorkspaceDirectory(operation, lister, request) _ = r.sendWorkspaceRead(context.WithoutCancel(ctx), env, result) }() return nil } -// workspaceResourceLocked returns what ref reads: the preparation or the run -// it admitted. -func (r *Router) workspaceResourceLocked(ref proto.AssignmentRef, request proto.WorkspaceReadPayload) (any, string) { +// workspaceResourceLocked returns what lists ref's directory: the local +// workspace, or without one the Harness Session of the Run ref admitted. +// Only the local workspace serves a preparation handle; a read-only +// preparation needs it. +func (r *Router) workspaceResourceLocked(ref proto.AssignmentRef, request proto.WorkspaceReadPayload) (agent.WorkspaceDirectoryLister, string) { if code := r.admitLocked(ref, ref.SessionID, request.EnvironmentID); code != "" { return nil, code } - var resource any if request.Handle != "" { p := r.preparations[request.Handle] if p == nil || p.request.Assignment != ref || p.environmentID != request.EnvironmentID || p.status.State != "ready" || - !p.owns || p.busy || p.ctx.Err() != nil || !time.Now().Before(p.deadline) { + !p.owns || p.busy || p.ctx.Err() != nil || !time.Now().Before(p.deadline) || r.localWorkspace == nil { return nil, "resource_unavailable" } - resource = p.prepared - if p.executor != nil { - resource = p.executor.native - } - } else { - s := r.sessions[request.RunID] - if s == nil || s.assignment != ref || s.environmentID != request.EnvironmentID || s.session == nil || - !r.interactionRouteOpenLocked(s) { - return nil, "resource_unavailable" - } - resource = s.session + return r.localWorkspace, "" + } + s := r.sessions[request.RunID] + if s == nil || s.assignment != ref || s.environmentID != request.EnvironmentID || s.session == nil || + !r.interactionRouteOpenLocked(s) { + return nil, "resource_unavailable" } if r.localWorkspace != nil { - resource = r.localWorkspace + return r.localWorkspace, "" } - return resource, "" + lister, ok := s.session.(agent.WorkspaceDirectoryLister) + if !ok { + return nil, "read_unsupported" + } + return lister, "" } -func executeWorkspaceRead(ctx context.Context, resource any, request proto.WorkspaceReadPayload) proto.WorkspaceReadResultPayload { - if request.Operation == "directory" { - reader, ok := resource.(agent.WorkspaceDirectoryLister) - if !ok { - return rejectedWorkspaceRead("read_unsupported") - } - read, err := reader.ListWorkspaceDirectory(ctx, request.Path, request.MaxEntries) - if err != nil { - return workspaceReadResult(agent.WorkspaceReadResult{}, err, 0) - } - if read.Entries == nil || len(read.Entries) > request.MaxEntries { - return workspaceReadResult(agent.WorkspaceReadResult{}, agent.ErrWorkspaceReadUncertain, 0) - } - directory := &proto.WorkspaceDirectoryResult{Entries: make([]proto.WorkspaceDirectoryEntry, 0, len(read.Entries)), Truncated: read.Truncated} - for _, entry := range read.Entries { - directory.Entries = append(directory.Entries, proto.WorkspaceDirectoryEntry{Name: entry.Name, Kind: entry.Kind, SizeBytes: entry.SizeBytes}) - } - if !proto.ValidWorkspaceDirectory(directory, request.MaxEntries) { - return workspaceReadResult(agent.WorkspaceReadResult{}, agent.ErrWorkspaceReadUncertain, 0) - } - return proto.WorkspaceReadResultPayload{Outcome: "completed", Directory: directory, CloseAcknowledged: true} +func listWorkspaceDirectory(ctx context.Context, lister agent.WorkspaceDirectoryLister, request proto.WorkspaceReadPayload) proto.WorkspaceReadResultPayload { + read, err := lister.ListWorkspaceDirectory(ctx, request.Path, request.MaxEntries) + if err != nil { + return workspaceReadFailure(err) } - reader, ok := resource.(agent.WorkspaceReader) - if !ok { - return rejectedWorkspaceRead("read_unsupported") + if read.Entries == nil || len(read.Entries) > request.MaxEntries { + return workspaceReadFailure(agent.ErrWorkspaceReadUncertain) } - read, err := reader.ReadWorkspaceFile(ctx, request.Path, request.MaxBytes) - return workspaceReadResult(read, err, request.MaxBytes) + directory := &proto.WorkspaceDirectoryResult{Entries: make([]proto.WorkspaceDirectoryEntry, 0, len(read.Entries)), Truncated: read.Truncated} + for _, entry := range read.Entries { + directory.Entries = append(directory.Entries, proto.WorkspaceDirectoryEntry{Name: entry.Name, Kind: entry.Kind, SizeBytes: entry.SizeBytes}) + } + if !proto.ValidWorkspaceDirectory(directory, request.MaxEntries) { + return workspaceReadFailure(agent.ErrWorkspaceReadUncertain) + } + return proto.WorkspaceReadResultPayload{Outcome: "completed", Directory: directory, CloseAcknowledged: true} } func rejectedWorkspaceRead(code string) proto.WorkspaceReadResultPayload { return proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: code} } -func workspaceReadResult(read agent.WorkspaceReadResult, err error, limit int) proto.WorkspaceReadResultPayload { - if err == nil && len(read.Data) <= limit && (!read.Truncated || len(read.Data) == limit) { - return proto.WorkspaceReadResultPayload{Outcome: "completed", Data: read.Data, Truncated: read.Truncated, CloseAcknowledged: true} - } +func workspaceReadFailure(err error) proto.WorkspaceReadResultPayload { for _, failure := range []struct { err error code string diff --git a/apps/daemon/internal/localworkspace/directory_native_test.go b/apps/daemon/internal/localworkspace/directory_native_test.go index ef22420f6..1aecaae54 100644 --- a/apps/daemon/internal/localworkspace/directory_native_test.go +++ b/apps/daemon/internal/localworkspace/directory_native_test.go @@ -33,24 +33,3 @@ func TestNativeDirectoryAPI(t *testing.T) { t.Fatal(got, err) } } - -func TestNativeFileReadPrefixAndEntryKinds(t *testing.T) { - b := nativeFileBinding(t) - if _, err := b.WriteWorkspaceFile(t.Context(), "nested/file", []byte{0, 255, 17}); err != nil { - t.Fatal(err) - } - got, err := b.ReadWorkspaceFile(t.Context(), "nested/file", 2) - if err != nil || !got.Truncated || len(got.Data) != 2 || got.Data[1] != 255 { - t.Fatal(got, err) - } - got, err = b.ReadWorkspaceFile(t.Context(), "nested/file", 3) - if err != nil || got.Truncated || len(got.Data) != 3 { - t.Fatal(got, err) - } - if _, err = b.ReadWorkspaceFile(t.Context(), "nested", 3); !errors.Is(err, agent.ErrWorkspaceReadInvalid) { - t.Fatal(err) - } - if _, err = b.ReadWorkspaceFile(t.Context(), "missing", 3); !os.IsNotExist(err) { - t.Fatal(err) - } -} diff --git a/apps/daemon/internal/localworkspace/native_files.go b/apps/daemon/internal/localworkspace/native_files.go index e669bc66a..bd1f2ac4d 100644 --- a/apps/daemon/internal/localworkspace/native_files.go +++ b/apps/daemon/internal/localworkspace/native_files.go @@ -277,53 +277,3 @@ func (r nativeContextReader) Read(data []byte) (int, error) { } return r.reader.Read(data) } - -// ReadWorkspaceFile returns the existing protocol's bounded prefix; the wire -// has no offset field. Non-regular entries cannot become unbounded streams. -func (b *Binding) ReadWorkspaceFile(ctx context.Context, path string, limit int) (agent.WorkspaceReadResult, error) { - result := agent.WorkspaceReadResult{} - if b == nil || ctx.Err() != nil { - return result, agent.ErrWorkspaceReadUnavailable - } - if path == "" || limit < 1 || limit > proto.WorkspaceReadMaxBytes { - return result, agent.ErrWorkspaceReadInvalid - } - local, err := nativeAPIPath(path) - if err != nil { - return result, agent.ErrWorkspaceReadInvalid - } - root, err := os.OpenRoot(b.workspace) - if err != nil { - return result, agent.ErrWorkspaceReadUnavailable - } - defer root.Close() - info, err := root.Lstat(local) - if err != nil { - return result, err - } - if !info.Mode().IsRegular() { - return result, agent.ErrWorkspaceReadInvalid - } - file, err := openNativePath(root, local) - if err != nil { - return result, err - } - defer file.Close() - info, err = file.Stat() - if err != nil { - return result, agent.ErrWorkspaceReadUncertain - } - if !info.Mode().IsRegular() { - return result, agent.ErrWorkspaceReadInvalid - } - raw, err := io.ReadAll(io.LimitReader(nativeContextReader{ctx: ctx, reader: file}, int64(limit)+1)) - if err != nil { - return result, agent.ErrWorkspaceReadUncertain - } - result.Truncated = len(raw) > limit - if result.Truncated { - raw = raw[:limit] - } - result.Data = raw - return result, nil -} diff --git a/contracts/agents-api/harness-onboarding.md b/contracts/agents-api/harness-onboarding.md index 38f679e45..c5cd4a2c3 100644 --- a/contracts/agents-api/harness-onboarding.md +++ b/contracts/agents-api/harness-onboarding.md @@ -70,7 +70,7 @@ For example, the Codex adapter keeps its app-server and thread, the Claude adapt | `Steerer` | Explicit implementation or Unsupported | Additional non-durable active-Turn input | | `FunctionResultSubmitter` | Explicit implementation or Unsupported | Match native call and result identity and acknowledge application | | `PermissionResponder`, `UserChoiceResponder` | Explicit implementation or Unsupported | Respond to exact emitted identities; unknown or expired interactions stay distinct from Unsupported | -| `WorkspaceReader`, `WorkspaceDirectoryLister`, `WorkspaceWriter` | Explicit on Turn and Executor owners | Use the authorized workspace, confirm access, commit or close, or return the operation's Unsupported error | +| `WorkspaceDirectoryLister`, `WorkspaceWriter` | Explicit on Turn and Executor owners | Use the authorized workspace, confirm access, commit or close, or return the operation's Unsupported error | | Neutral messages, images, MCP, structured output and Subagent observations | Explicit capability decisions | Keep each operation's protocol semantics; reject unsupported input before submission | Each adapter's `contracts.go` holds an individual compile-time assertion for each small interface. Do not embed a default implementation that makes future interfaces appear implemented. Adding a contract also requires a classification in the common completeness check and an explicit assertion in every public adapter; the check follows the authored Harness catalog. diff --git a/contracts/agents-api/zh/harness-onboarding.md b/contracts/agents-api/zh/harness-onboarding.md index 7309f4843..fc0fccdca 100644 --- a/contracts/agents-api/zh/harness-onboarding.md +++ b/contracts/agents-api/zh/harness-onboarding.md @@ -1,7 +1,7 @@ --- title: "将原生 Harness 添加到 OpenAgentCore" source: contracts/agents-api/harness-onboarding.md -source_hash: d0b84eef91e766fb74e2acf8a767d1767f6a3f270adb1b72790b7c616ee63d05 +source_hash: 6a046c227cb9d0b3074a21c9257a408c126234fd127ad08f359d89891ee0f20b --- **Harness** 是一种运行模型和工具循环的原生代理引擎（Codex、Claude Code、MiniMax Code）。**Harness 适配器**将 Runtime 的 Executor 和 Turn 契约转换到该引擎的 SDK 或协议。本文档定义 Runtime–Harness 协议：适配器接口及其生命周期义务、注册、Core 资格认定和验收。[Harness capabilities](harness-capabilities.md) 记录了当前每个 Harness 支持的功能。 @@ -72,7 +72,7 @@ Environment 提供执行资源。受管 E2B、Docker 和 microsandbox 机器以 | `Steerer` | 明确实现或 Unsupported | 额外的非持久化活动 Turn 输入 | | `FunctionResultSubmitter` | 明确实现或 Unsupported | 匹配原生调用和结果身份，并确认应用 | | `PermissionResponder`、`UserChoiceResponder` | 明确实现或 Unsupported | 响应精确发出的身份；未知或已过期的交互与 Unsupported 保持区分 | -| `WorkspaceReader`、`WorkspaceDirectoryLister`、`WorkspaceWriter` | 在 Turn 和 Executor 所有者上明确实现 | 使用授权工作区，确认访问，提交或关闭，或者返回该操作的 Unsupported 错误 | +| `WorkspaceDirectoryLister`、`WorkspaceWriter` | 在 Turn 和 Executor 所有者上明确实现 | 使用授权工作区，确认访问，提交或关闭，或者返回该操作的 Unsupported 错误 | | 中立消息、图像、MCP、结构化输出和 Subagent 观察 | 明确作出能力决策 | 保持每项操作的协议语义；在提交前拒绝不受支持的输入 | 每个适配器的 `contracts.go` 都包含针对每个小型接口的单项编译时断言。不要嵌入会让未来接口看起来已经实现的默认实现。添加契约时，还必须在通用完整性检查中进行分类，并在每个公共适配器中添加明确断言；该检查遵循已编写的 Harness 目录。 diff --git a/docs/runtime-protocol.md b/docs/runtime-protocol.md index 01c644f7e..97fef0b42 100644 --- a/docs/runtime-protocol.md +++ b/docs/runtime-protocol.md @@ -202,13 +202,13 @@ Core stores accepted values in the Turn outcome as `engine_error_code` and `engi A workspace read that needs no running Turn uses the read-only preparation profile: `execution_prepare` with `workspace_read_only`, which requires the `workspace_read_preparation` capability. It accepts only the bound Environment and resource identity; execution options, model and MCP credentials, native Session continuation and model or tool input are excluded, and the owner rejects `execution_start`. The Runtime serves it from its bound local workspace without starting a Harness process. The profile publishes `released` only after local close succeeds; a cleanup error keeps ownership and reports `cleanup_unconfirmed`. A successful cleanup retry publishes the confirmed release; a stale status snapshot never publishes success. A release request, HTTP disconnect or remote socket closure alone does not confirm cleanup. -`workspace_read` targets an existing preparation handle, or the Run it was transferred to, on the same authenticated device connection, with the exact frozen Environment identity; callers cannot supply sockets, credentials or workspace roots. `operation: directory` lists one workspace-relative directory (an empty path selects the root) with mutually exclusive byte and entry limits. A result carries at most 1024 single-component UTF-8 names of at most 255 bytes each, the entry kind, regular-file sizes and explicit truncation, and is returned only after directory access and handle cleanup settle. There is no snapshot, recursion or pagination at this layer. Byte and directory reads share target checks, correlation, capacity and retained operation waits. +`workspace_read` lists one workspace-relative directory (an empty path selects the root) of an existing preparation handle, or the Run it was transferred to, on the same authenticated device connection, with the exact frozen Environment identity; callers cannot supply sockets, credentials or workspace roots. A result carries at most `max_entries` (1 to 1024) single-component UTF-8 names of at most 255 bytes each, the entry kind, regular-file sizes and explicit truncation, and is returned only after directory access and handle cleanup settle. There is no snapshot, recursion or pagination at this layer. -Request payloads are bounded at 8 KiB and correlation IDs at 128 bytes before admission; oversized IDs are not echoed, and oversized trace metadata is omitted from replies. Raw read results are bounded at 1 MiB within the 4 MiB transport frame. These are private transport limits, not public Files parameters. A successful read requires complete bytes or explicit truncation and an acknowledged native close. A safe native rejection carries no bytes; an interrupted or ambiguous read stays unknown and stops further reads on that owner. A dispatched read keeps its bounded waiter across observer cancellation and resource transfer or release, and resource closure stops new admission. The gateway bounds subscriptions and never retries or replays a read on reconnect; a duplicate pending operation ID cannot start another read. +Request payloads are bounded at 8 KiB and correlation IDs at 128 bytes before admission; oversized IDs are not echoed, and oversized trace metadata is omitted from replies. These are private transport limits, not public Files parameters. A safe native rejection carries no directory; an interrupted or ambiguous read stays unknown and stops further reads on that owner. A dispatched read keeps its bounded waiter across observer cancellation and resource transfer or release, and resource closure stops new admission. The gateway bounds subscriptions and never retries or replays a read on reconnect; a duplicate pending operation ID cannot start another read. Core runs an idle directory read on the Worker's Session scheduling reservation and targets the exact Run during active execution. It keeps the reservation through the bounded read and release, returns data only after a confirmed close (an incomplete read or uncertain cleanup returns unavailable without data), releases the reservation before delivering the result, and revokes the scoped read credential on completion or failure. The Runtime keeps uncertain cleanup ownership and capacity. The [Environment Files contract](../contracts/agents-api/environment-files.md) owns public authorization, paths and pagination. -`workspace_write` transfers a complete bounded body in acknowledged 64 KiB frames before the native writer runs, verifies the declared digest and runs no model. The private transfer bound is 50 MiB, separate from the public 5 MiB decoded inline bound that the API checks before any Runtime work. The Runtime excludes execution while it receives or applies a write; a malformed, incomplete or expired transfer never reaches the installer. An exact commit or rejection receipt releases the mutation owner. A missing or ambiguous receipt keeps the uncertainty: observer cancellation and local process exit cannot prove that nothing changed. Before public admission Core durably reserves the write under the Session lock and blocks successor mutations across restarts until exact settlement; the request is never replayed. Every platform uses the daemon's Go implementation for bounded reads, directory listing, file creation and output export, with no external helper or staging directory. +`workspace_write` transfers a complete bounded body in acknowledged 64 KiB frames before the native writer runs, verifies the declared digest and runs no model. The private transfer bound is 50 MiB, separate from the public 5 MiB decoded inline bound that the API checks before any Runtime work. The Runtime excludes execution while it receives or applies a write; a malformed, incomplete or expired transfer never reaches the installer. An exact commit or rejection receipt releases the mutation owner. A missing or ambiguous receipt keeps the uncertainty: observer cancellation and local process exit cannot prove that nothing changed. Before public admission Core durably reserves the write under the Session lock and blocks successor mutations across restarts until exact settlement; the request is never replayed. Every platform uses the daemon's Go implementation for directory listing, file creation and output export, with no external helper or staging directory. ## MCP connection authority diff --git a/docs/zh/runtime-protocol.md b/docs/zh/runtime-protocol.md index 2def15eb2..4b9131b00 100644 --- a/docs/zh/runtime-protocol.md +++ b/docs/zh/runtime-protocol.md @@ -1,7 +1,7 @@ --- title: "Core–Runtime 协议" source: docs/runtime-protocol.md -source_hash: 9ef338b2da6b53a2d5290527ebff5bb119c8c85f4fdeef8dd2cf8bde3ec16e0d +source_hash: 5f4fc0a408f12938550bb548f6b95c336316be2320f5cbb8c6801a823d88863e --- 此协议在 Runtime daemon 获取机器凭据后连接 Core 与 daemon，定义 daemon 连接上消息的含义和顺序。wire 类型、限制和验证器仅在 [`internal/agentdaemon/proto`](https://github.com/MiniMax-AI/OpenAgentCore/tree/main/internal/agentdaemon/proto) 中定义一次；Core 的 [gateway](https://github.com/MiniMax-AI/OpenAgentCore/tree/main/services/core/internal/runtimegateway) 与参考 Runtime 的 [dispatcher](https://github.com/MiniMax-AI/OpenAgentCore/tree/main/apps/daemon/internal/dispatch) 都使用它们，因此无需同步第二套 payload schema。签发凭据和打开连接的 HTTP 路由见[机器连接 API](../../contracts/agents-api/zh/machine-api.md)。 @@ -204,13 +204,13 @@ Core 在 Turn outcome 中将接受的值保存为 `engine_error_code` 和 `engin 无需运行 Turn 的工作区读取使用只读 preparation profile：带 `workspace_read_only` 的 `execution_prepare`，要求 `workspace_read_preparation` 能力。仅接受绑定的 Environment 和 resource 身份；不包含 execution option、model 与 MCP 凭据、原生 Session continuation、model 或 tool 输入，owner 拒绝 `execution_start`。Runtime 从绑定的本地工作区提供读取，不启动 Harness 进程。profile 仅在本地 close 成功后发布 `released`；清理错误保留所有权并报告 `cleanup_unconfirmed`。清理重试成功后发布已确认释放；旧 status snapshot 不发布成功。release 请求、HTTP 断连或远端 socket 关闭本身都不确认清理。 -`workspace_read` 在同一已认证设备连接上，针对现有 preparation handle 或它已转移给的 Run，使用精确冻结的 Environment 身份；调用方不能提供 socket、凭据或 workspace root。`operation: directory` 列出一个 workspace 相对目录（空路径选择根目录），字节与条目限制互斥。结果最多携带 1024 个单路径组件 UTF-8 名称，每个最多 255 字节，并包含 entry kind、普通文件大小和明确截断信息；仅在目录访问与 handle 清理结算后返回。此层没有快照、递归或分页。字节读取与目录读取共享目标检查、关联、容量和保留的操作等待。 +`workspace_read` 在同一已认证设备连接上，针对现有 preparation handle 或它已转移给的 Run，使用精确冻结的 Environment 身份，列出一个 workspace 相对目录（空路径选择根目录）；调用方不能提供 socket、凭据或 workspace root。结果最多携带 `max_entries`（1 到 1024）个单路径组件 UTF-8 名称，每个最多 255 字节，并包含 entry kind、普通文件大小和明确截断信息；仅在目录访问与 handle 清理结算后返回。此层没有快照、递归或分页。 -准入前，请求 payload 限制为 8 KiB，correlation ID 限制为 128 字节；过长 ID 不回显，过大的 trace metadata 不进入回复。原始 read result 限制为 1 MiB，位于 4 MiB transport frame 内。这些是私有 transport 限制，不是公开 Files 参数。成功读取要求完整字节或明确截断，并确认原生 close。安全的原生拒绝不携带字节；中断或有歧义的读取保持未知，并停止该 owner 的后续读取。已 dispatch 的读取在 observer 取消以及资源转移或释放后仍保留有限时等待方，资源关闭阻止新准入。gateway 限制订阅，不在重连后重试或重放读取；重复的 pending operation ID 不能启动另一次读取。 +准入前，请求 payload 限制为 8 KiB，correlation ID 限制为 128 字节；过长 ID 不回显，过大的 trace metadata 不进入回复。这些是私有 transport 限制，不是公开 Files 参数。安全的原生拒绝不携带目录；中断或有歧义的读取保持未知，并停止该 owner 的后续读取。已 dispatch 的读取在 observer 取消以及资源转移或释放后仍保留有限时等待方，资源关闭阻止新准入。gateway 限制订阅，不在重连后重试或重放读取；重复的 pending operation ID 不能启动另一次读取。 Core 在 Worker 的 Session 调度预约上运行空闲目录读取，活动执行时针对精确 Run。它在有限时 read 与 release 期间保留预约，仅在确认 close 后返回数据（不完整读取或不确定清理返回 unavailable，不包含数据），在交付结果前释放预约，并在完成或失败后撤销限定作用域的读取凭据。Runtime 保留不确定清理的所有权和容量。[Environment Files 契约](../../contracts/agents-api/zh/environment-files.md)负责公开授权、路径和分页。 -`workspace_write` 在原生 writer 运行前，通过已确认的 64 KiB frame 传输完整且有界的 body，验证声明的 digest，不运行模型。私有 transfer 限制为 50 MiB，与公开 API 在任何 Runtime 工作前检查的 5 MiB decoded inline 限制独立。Runtime 在接收或应用写入时排除执行；格式错误、不完整或到期的 transfer 不会到达 installer。精确的 commit 或拒绝回执释放 mutation owner。缺失或有歧义的回执保留不确定性：observer 取消和本地进程退出不能证明没有改变任何内容。公开准入前，Core 在 Session lock 下持久预约写入，跨重启阻止后继 mutation，直到精确结算；请求不重放。各平台都使用 daemon 的 Go 实现进行有界读取、目录列举、文件创建与输出导出，不使用外部 helper 或 staging directory。 +`workspace_write` 在原生 writer 运行前，通过已确认的 64 KiB frame 传输完整且有界的 body，验证声明的 digest，不运行模型。私有 transfer 限制为 50 MiB，与公开 API 在任何 Runtime 工作前检查的 5 MiB decoded inline 限制独立。Runtime 在接收或应用写入时排除执行；格式错误、不完整或到期的 transfer 不会到达 installer。精确的 commit 或拒绝回执释放 mutation owner。缺失或有歧义的回执保留不确定性：observer 取消和本地进程退出不能证明没有改变任何内容。公开准入前，Core 在 Session lock 下持久预约写入，跨重启阻止后继 mutation，直到精确结算；请求不重放。各平台都使用 daemon 的 Go 实现进行目录列举、文件创建与输出导出，不使用外部 helper 或 staging directory。 ## MCP 连接权限 {#mcp-connection-authority} diff --git a/internal/agentdaemon/proto/workspace_directory.go b/internal/agentdaemon/proto/workspace_directory.go index 4209bf345..d11e3bf3c 100644 --- a/internal/agentdaemon/proto/workspace_directory.go +++ b/internal/agentdaemon/proto/workspace_directory.go @@ -36,17 +36,8 @@ func (result *WorkspaceDirectoryResult) UnmarshalJSON(data []byte) error { } func ValidWorkspaceReadRequest(request WorkspaceReadPayload) bool { - if (request.Handle == "") == (request.RunID == "") || request.EnvironmentID == "" { - return false - } - switch request.Operation { - case "": - return request.MaxBytes >= 1 && request.MaxBytes <= WorkspaceReadMaxBytes && request.MaxEntries == 0 - case "directory": - return request.MaxBytes == 0 && request.MaxEntries >= 1 && request.MaxEntries <= WorkspaceDirectoryMaxEntries - default: - return false - } + return (request.Handle == "") != (request.RunID == "") && request.EnvironmentID != "" && + request.MaxEntries >= 1 && request.MaxEntries <= WorkspaceDirectoryMaxEntries } func ValidWorkspaceDirectory(result *WorkspaceDirectoryResult, limit int) bool { diff --git a/internal/agentdaemon/proto/workspace_read.go b/internal/agentdaemon/proto/workspace_read.go index 428224554..63e640bcd 100644 --- a/internal/agentdaemon/proto/workspace_read.go +++ b/internal/agentdaemon/proto/workspace_read.go @@ -3,7 +3,6 @@ package proto const ( TypeWorkspaceRead = "workspace_read" TypeWorkspaceReadResult = "workspace_read_result" - WorkspaceReadMaxBytes = 1 << 20 WorkspaceReadMaxRequestBytes = 8 << 10 WorkspaceReadMaxIDBytes = 128 // WorkspaceReadNotDirectory rejects a directory read whose own path is @@ -11,22 +10,18 @@ const ( WorkspaceReadNotDirectory = "not_directory" ) -// WorkspaceReadPayload targets one existing resource on the current daemon connection. +// WorkspaceReadPayload lists one directory of an existing resource on the current daemon connection. type WorkspaceReadPayload struct { Handle string `json:"handle,omitempty"` RunID string `json:"run_id,omitempty"` EnvironmentID string `json:"environment_id"` Path string `json:"path"` - MaxBytes int `json:"max_bytes"` - Operation string `json:"operation,omitempty"` MaxEntries int `json:"max_entries,omitempty"` } -// WorkspaceReadResultPayload never infers file settlement from local process exit. +// WorkspaceReadResultPayload never infers settlement from local process exit. type WorkspaceReadResultPayload struct { Outcome string `json:"outcome"` - Data []byte `json:"data,omitempty"` - Truncated bool `json:"truncated,omitempty"` CloseAcknowledged bool `json:"close_acknowledged,omitempty"` ErrorCode string `json:"error_code,omitempty"` Directory *WorkspaceDirectoryResult `json:"directory,omitempty"` diff --git a/packages/claude-sdk-adapter/README.md b/packages/claude-sdk-adapter/README.md index 329b832b9..2b374732b 100644 --- a/packages/claude-sdk-adapter/README.md +++ b/packages/claude-sdk-adapter/README.md @@ -38,13 +38,11 @@ Each Turn ends with a result/error and `turn_settled`, independently of process `claudesdk.NewExecutorFactory` binds this bridge to `agent.Executor`. Direct-call and read-only preparation wrappers delegate to the same implementation. Runtime execution uses the Executor registry for both none and workspace configurations. Its owner context spans all Turns; a Turn's caller cannot replace fixed resources. A failed preparation returns its Executor when cleanup remains unconfirmed. Installed runtime checks are cached by package/file identity, while capability and request validation still run for each Executor configuration. -### Workspace reads and directory listing - -The optional private `agent.WorkspaceReader` on this Executor and its delegated wrappers requires the packaged `workspace_read` feature. It sends bounded relative paths to that same SDK Query's native `readFile` control. Only the adapter combines the path with the frozen workspace root; callers cannot replace the placement. The pinned native read handler awaits file-handle close before its successful base64 response. The adapter validates bytes and truncation, bounds each result to 1 MiB and each request to 8 KiB, and admits one read at a time. Its continuous bridge output consumer retains read receipts during preparation and across Turns. Caller cancellation detaches observation without cancelling the Run or discarding an admitted waiter; its original deadline still applies. Owner closure stops admission. Native null, malformed receipts, timeout and interrupted delivery remain uncertain and stop the owner; local reap is not a successful read settlement. The SDK's nullable result catches all native/control errors, so it cannot distinguish missing files from denial or transport failure. The reader provides no public Files endpoint, snapshot consistency, placement registration or idle owner policy. The qualified live workspace fixture also checks binary, empty and bounded reads before input and during real execution, plus effects before cancellation and reads on fresh-process history continuation. +### Directory listing The optional private `agent.WorkspaceDirectoryLister` uses the portable `workspace_directory` bridge feature. Node reads directory metadata under the selected workspace; there is no Linux `/proc/self/fd` dependency. Public daemon Files operations share the Go Binding implementation on all platforms. Their relative path and result bounds define the Files API, not Harness permissions. -Directory requests are bounded to 8 KiB and 1,000 immediate entries, with explicit truncation, literal names, kinds, and sizes only for regular files. They do not promise ordering, snapshots, recursion, or public pagination. Missing and permission errors are returned only from distinguishable filesystem outcomes; unknown results stop the owner. Each operation closes its directory before a successful receipt. Caller cancellation, Turn transitions and owner shutdown follow the workspace read settlement rules. The lister alone does not enable public Claude Files; the dedicated Runtime integration supplies public placement and ownership. +Directory requests are bounded to 8 KiB and 1,000 immediate entries, with explicit truncation, literal names, kinds, and sizes only for regular files. They do not promise ordering, snapshots, recursion, or public pagination. Missing and permission errors are returned only from distinguishable filesystem outcomes; unknown results stop the owner. Each operation closes its directory before a successful receipt. The continuous bridge output consumer retains receipts during preparation and across Turns. Caller cancellation detaches observation without cancelling the Run or discarding an admitted waiter, whose original deadline still applies; owner closure stops admission. The lister alone does not enable public Claude Files; the dedicated Runtime integration supplies public placement and ownership. ### Command observations diff --git a/packages/claude-sdk-adapter/src/adapter.ts b/packages/claude-sdk-adapter/src/adapter.ts index e0375740f..a1e8ccd75 100644 --- a/packages/claude-sdk-adapter/src/adapter.ts +++ b/packages/claude-sdk-adapter/src/adapter.ts @@ -6,7 +6,6 @@ import { Subagents } from "./subagents.js"; import type { Fact } from "./subagent_history.js"; import { WorkspaceDirectories, type WorkspaceDirectoryEvent } from "./workspace_directories.js"; import { getSessionInfo, query, startup, type McpServerConfig, type Options, type WarmQuery } from "@anthropic-ai/claude-agent-sdk"; -import { WorkspaceReads, type WorkspaceReadEvent } from "./workspace_reads.js"; import { Inputs, type InputEvent } from "./inputs.js"; import { executorResultUsage, resultUsage, type NativeUsage } from "./usage.js"; import { spawnNative } from "./native.js"; @@ -25,7 +24,6 @@ export type Event = | ExecutorEvent | Fact | WorkspaceDirectoryEvent - | WorkspaceReadEvent | MessageEvent | InputEvent | FunctionEvent @@ -37,7 +35,7 @@ export type Event = | { type: "result"; session_id: string; text: string } | { type: "error"; code: "invalid_request" | "history_unavailable" | "execution_failed" | "cancelled"; engine_error_code?: string; session_id?: string; result_id?: string }; -export async function execute(request: Start | Prepare | ExecutorPrepare, emit: (event: Event) => Promise, abort: AbortController, functions = new FunctionBridge(emit), inputs = new Inputs(immediateInput(request)), reads = new WorkspaceReads(emit, abort), directories = new WorkspaceDirectories(emit, abort), turns?: ExecutorTurns): Promise { +export async function execute(request: Start | Prepare | ExecutorPrepare, emit: (event: Event) => Promise, abort: AbortController, functions = new FunctionBridge(emit), inputs = new Inputs(immediateInput(request)), directories = new WorkspaceDirectories(emit, abort), turns?: ExecutorTurns): Promise { const ownerEmit=emit; if(turns) emit=event=>turns.emit(event); const nativeFailure = new NativeFailure(); @@ -137,10 +135,7 @@ export async function execute(request: Start | Prepare | ExecutorPrepare, emit: if ((hooksRequested && initialized.hooks_applied !== true) || children.length !== 1 || !nativeAlive || abort.signal.aborted) { throw new Error("preparation unavailable"); } - if(workspace) { - reads.bind(stream, request.cwd); - if (process.platform === "linux") await directories.bind(request.cwd); - } + if (workspace && process.platform === "linux") await directories.bind(request.cwd); if (declarations?.some(server => "required" in server && server.required)) profile?.verifyRequired(await stream.mcpServerStatus()); if(turns) { turns.configure(stream,(input,output)=>{ @@ -233,7 +228,6 @@ export async function execute(request: Start | Prepare | ExecutorPrepare, emit: turns?.close(); functions.close(); try { await directories.close(); } catch { failed = true; classifiedFailure = undefined; } - await reads.close(); stream?.close(); warm?.close(); abort.signal.removeEventListener("abort", closeInputs); diff --git a/packages/claude-sdk-adapter/src/main.ts b/packages/claude-sdk-adapter/src/main.ts index 86b12aa09..e3ba0bd11 100644 --- a/packages/claude-sdk-adapter/src/main.ts +++ b/packages/claude-sdk-adapter/src/main.ts @@ -1,6 +1,5 @@ import { ExecutorTurns } from "./executor_protocol.js"; import { WorkspaceDirectories } from "./workspace_directories.js"; -import { WorkspaceReads } from "./workspace_reads.js"; import { Inputs } from "./inputs.js"; import { FunctionBridge } from "./function_bridge.js"; import { createInterface } from "node:readline"; @@ -28,7 +27,6 @@ try { await emit(invalid && (event.type === "error" || event.type === "result") ? { type: "error", code: "invalid_request" } : event); }; const functions = new FunctionBridge(output); - const reads = new WorkspaceReads(output, abort); const directories = new WorkspaceDirectories(output, abort); const prompts = new Inputs(immediateInput(request)); const turns=request.type==="executor_prepare" ? new ExecutorTurns(emit,abort) : undefined; @@ -37,9 +35,7 @@ try { for await (const line of { [Symbol.asyncIterator]: () => input }) { if (Buffer.byteLength(line) > 1024 * 1024) throw new Error("Invalid input."); const value: unknown = JSON.parse(line); - if (value && typeof value === "object" && "type" in value && value.type === "workspace_read") { - reads.submit(value as Record); - } else if (value && typeof value === "object" && "type" in value && value.type === "workspace_directory") { + if (value && typeof value === "object" && "type" in value && value.type === "workspace_directory") { directories.submit(value as Record); } else if(turns) { if(!value || typeof value!=="object" || Array.isArray(value)) throw new Error("invalid_request"); @@ -58,7 +54,7 @@ try { } catch { invalid = request.type === "prepare"; abort.abort(); } })(); - try { await execute(request, output, abort, functions, prompts, reads, directories, turns); } + try { await execute(request, output, abort, functions, prompts, directories, turns); } catch { await output({ type: "error", code: "execution_failed" }); } finally { prompts.close(); diff --git a/packages/claude-sdk-adapter/src/runtime_check.ts b/packages/claude-sdk-adapter/src/runtime_check.ts index c20f3eab8..b8a4a1852 100644 --- a/packages/claude-sdk-adapter/src/runtime_check.ts +++ b/packages/claude-sdk-adapter/src/runtime_check.ts @@ -45,7 +45,7 @@ try { assert.equal(smoke.error, undefined, "bridge_unavailable"); assert.equal(smoke.status, 0, "bridge_unavailable"); assert.deepEqual(JSON.parse(smoke.stdout), { type: "error", code: "invalid_request" }); - process.stdout.write(JSON.stringify({ type: "runtime_ready", protocol: 3, features: ["executor_reuse", ...(["linux", "darwin", "win32"].includes(process.platform) ? ["workspace_directory", "local_runtime_v2", "workspace_functions", "workspace_structured_output", "workspace_tool_search", "workspace_mcp_http"] : []), "message_images", "function_result_images", "tool_search", "structured_output", "subagent_resources", "mcp_http_tools", "mcp_http_bearer_auth", "mcp_http_required", "workspace_tools", "workspace_prepare", "workspace_read", "workspace_command_observations"], node: process.versions.node, sdk: sdk.version, mcp: mcp.version, native: nativeVersion, native_path: relative(root, binary) }) + "\n"); + process.stdout.write(JSON.stringify({ type: "runtime_ready", protocol: 3, features: ["executor_reuse", ...(["linux", "darwin", "win32"].includes(process.platform) ? ["workspace_directory", "local_runtime_v2", "workspace_functions", "workspace_structured_output", "workspace_tool_search", "workspace_mcp_http"] : []), "message_images", "function_result_images", "tool_search", "structured_output", "subagent_resources", "mcp_http_tools", "mcp_http_bearer_auth", "mcp_http_required", "workspace_tools", "workspace_prepare", "workspace_command_observations"], node: process.versions.node, sdk: sdk.version, mcp: mcp.version, native: nativeVersion, native_path: relative(root, binary) }) + "\n"); } catch { // Native diagnostics can include operator environment; never forward them. process.stdout.write(JSON.stringify({ type: "runtime_unavailable" }) + "\n"); diff --git a/packages/claude-sdk-adapter/src/workspace_reads.ts b/packages/claude-sdk-adapter/src/workspace_reads.ts deleted file mode 100644 index fc9c6aa4e..000000000 --- a/packages/claude-sdk-adapter/src/workspace_reads.ts +++ /dev/null @@ -1,57 +0,0 @@ -import type { Query } from "@anthropic-ai/claude-agent-sdk"; -import { join } from "node:path"; - -export type WorkspaceReadEvent = { type: "workspace_read"; id: string } & ( - { data_base64: string; truncated: boolean } | { error: "invalid" | "busy" | "unavailable" | "uncertain" } -); - -export class WorkspaceReads { - private query?: Pick; - private root = ""; - private stopped = false; - private pending?: Promise; - - constructor(private readonly emit: (event: WorkspaceReadEvent) => Promise, private readonly abort: AbortController) {} - - bind(query: Pick, root: string): void { this.query = query; this.root = root; } - - submit(value: Record): void { - if (typeof value.id !== "string" || !/^[a-zA-Z0-9-]{1,128}$/.test(value.id)) throw new Error("invalid_request"); - const id = value.id; - let error: "invalid" | "busy" | "unavailable" | undefined; - if (Buffer.byteLength(JSON.stringify(value)) > 8192 || Object.keys(value).some(key => !["type", "id", "path", "max_bytes"].includes(key)) || - typeof value.path !== "string" || !value.path || /[\x00\\\r\n]/.test(value.path) || value.path.split("/").some(part => !part || part === "." || part === "..") || - !Number.isInteger(value.max_bytes) || (value.max_bytes as number) < 1 || (value.max_bytes as number) > 1048576) error = "invalid"; - else if (this.stopped || this.abort.signal.aborted || !this.query) error = "unavailable"; - else if (this.pending) error = "busy"; - if (error) { void this.emit({ type: "workspace_read", id, error }).catch(() => this.abort.abort()); return; } - const query = this.query!; - const path = join(this.root, value.path as string); - const maxBytes = value.max_bytes as number; - this.pending = this.read(query, id, path, maxBytes).finally(() => { this.pending = undefined; }); - } - - private async read(query: Pick, id: string, path: string, maxBytes: number): Promise { - let timer: ReturnType | undefined; - const native = query.readFile(path, { maxBytes, encoding: "base64" }); - try { - const result = await Promise.race([native, new Promise((_, reject) => { - timer = setTimeout(() => reject(new Error("native read timeout")), 10000); - })]); - if (!result || result.encoding !== "base64" || result.absPath !== path || typeof result.contents !== "string" || - (result.truncated !== undefined && typeof result.truncated !== "boolean") || result.contents.length > 4 * Math.ceil(maxBytes / 3)) throw new Error("uncertain"); - const bytes = Buffer.from(result.contents, "base64"); - if (bytes.toString("base64") !== result.contents || bytes.length > maxBytes || (result.truncated && bytes.length !== maxBytes)) throw new Error("uncertain"); - await this.emit({ type: "workspace_read", id, data_base64: result.contents, truncated: result.truncated === true }); - } catch { - this.stopped = true; - await this.emit({ type: "workspace_read", id, error: "uncertain" }).catch(() => {}); - this.abort.abort(); - // SDK close resolves its retained request waiter; timeout alone is not native settlement. - query.close(); - await native.catch(() => null); - } finally { clearTimeout(timer); } - } - - async close(): Promise { this.stopped = true; await this.pending; } -} diff --git a/packages/claude-sdk-adapter/tests/preparation.test.mjs b/packages/claude-sdk-adapter/tests/preparation.test.mjs index 8b7a4a590..fb3e4fdc0 100644 --- a/packages/claude-sdk-adapter/tests/preparation.test.mjs +++ b/packages/claude-sdk-adapter/tests/preparation.test.mjs @@ -40,13 +40,6 @@ globalThis.startupFixture = async ({options, initializeTimeoutMs}) => { if(mode === "native-exit")setTimeout(()=>child.kill("SIGTERM"),100); return { close, - async readFile(path,{maxBytes,encoding}) { - assert.equal(encoding,"base64"); - assert.equal(path,options.cwd+"/binary"); - process.send({kind:"read",path,maxBytes}); - await new Promise(resolve=>setTimeout(resolve,80)); - return {absPath:path,encoding,contents:Buffer.from([0,255,128,1]).toString("base64")}; - }, async initializationResult(){return {hooks_applied:mode !== "missing-hooks"};}, async *[Symbol.asyncIterator](){ const command={type:"assistant",session_id:"native",parent_tool_use_id:null, @@ -205,16 +198,3 @@ for (const mode of ["missing-history", "missing-hooks", "startup-error", "early- if (mode === "missing-history") assert.equal(observations.some(value => value.kind === "spawn"), false); }); } - - -test("prepared native reader stays on the same query across Start", { timeout: 10000 }, async t => { - const { request, events, observations, send, wait, finish } = await launch(t, "release"); - send(request); await wait(() => events.some(event => event.type === "prepared")); - send({type:"workspace_read",id:"before",path:"binary",max_bytes:4}); - await wait(()=>observations.some(value=>value.kind === "read")); - send({type:"start",input: [{ content: [{ type: "input_text", text: "first" }] }]}); - await wait(()=>events.some(event=>event.type === "workspace_read")); - assert.deepEqual(events.find(event=>event.type === "workspace_read"),{type:"workspace_read",id:"before",data_base64:"AP+AAQ==",truncated:false}); - await finish(); - assert.equal(observations.filter(value=>value.kind === "spawn").length,1); -}); diff --git a/packages/claude-sdk-adapter/tests/workspace_reads.test.mjs b/packages/claude-sdk-adapter/tests/workspace_reads.test.mjs deleted file mode 100644 index 5d8fe943c..000000000 --- a/packages/claude-sdk-adapter/tests/workspace_reads.test.mjs +++ /dev/null @@ -1,34 +0,0 @@ -import assert from "node:assert/strict"; -import test from "node:test"; -import { WorkspaceReads } from "../dist/workspace_reads.js"; - -const request = { type: "workspace_read", id: "read", path: "file", max_bytes: 4 }; -function fixture(readFile) { - const events=[],abort=new AbortController();let closed=false; - const reads=new WorkspaceReads(async event=>{events.push(event);},abort); - reads.bind({readFile,close(){closed=true;}},"/workspace"); - return {reads,events,abort,get closed(){return closed;}}; -} -for (const [name,contents,truncated] of [["binary","AP+AAQ==",false],["empty","",false],["prefix","AP+AAQ==",true]]) { - test(`native binary response: ${name}`,async()=>{ - const f=fixture(async(path,options)=>{assert.equal(path,"/workspace/file");assert.deepEqual(options,{maxBytes:4,encoding:"base64"});return {absPath:path,encoding:"base64",contents,...truncated&&{truncated}};}); - f.reads.submit(request);await f.reads.close(); - assert.deepEqual(f.events,[{type:"workspace_read",id:"read",data_base64:contents,truncated}]);assert.equal(f.abort.signal.aborted,false); - }); -} -test("bounds, admission and release retain the original native wait",async()=>{ - let settle;const f=fixture(()=>new Promise(resolve=>{settle=resolve;})); - for(const fields of [{path:"../escape"},{path:"/absolute"},{path:"a//b"},{path:"a\\b"},{max_bytes:0},{max_bytes:1048577},{path:"a".repeat(8192)}]) f.reads.submit({...request,...fields}); - assert.ok(f.events.every(event=>event.error === "invalid"));f.events.length=0; - f.reads.submit(request);f.reads.submit({...request,id:"busy"});assert.equal(f.events[0].error,"busy"); - let released=false;const release=f.reads.close().then(()=>{released=true;});await Promise.resolve();assert.equal(released,false); - f.reads.submit({...request,id:"closed"});assert.equal(f.events[1].error,"unavailable"); - settle({absPath:"/workspace/file",encoding:"base64",contents:""});await release;assert.equal(f.events[2].data_base64,""); -}); -for(const result of [null,{absPath:"/workspace/file",contents:"",encoding:"utf-8"},{absPath:"/other",contents:"",encoding:"base64"}, - {absPath:"/workspace/file",contents:"!",encoding:"base64"},{absPath:"/workspace/file",contents:"",encoding:"base64",truncated:true}]) { - test("ambiguous native responses fence the owner",async()=>{ - const f=fixture(async()=>result);f.reads.submit(request);await f.reads.close(); - assert.deepEqual(f.events,[{type:"workspace_read",id:"read",error:"uncertain"}]);assert.equal(f.abort.signal.aborted,true);assert.equal(f.closed,true); - }); -} diff --git a/scripts/check-claude-sdk-runtime.mjs b/scripts/check-claude-sdk-runtime.mjs index 9675881e2..c6aee4f11 100644 --- a/scripts/check-claude-sdk-runtime.mjs +++ b/scripts/check-claude-sdk-runtime.mjs @@ -23,7 +23,7 @@ assert.equal(probe.status, 0, "Exported runtime is unavailable"); const report = JSON.parse(probe.stdout); assert.equal(report.type, "runtime_ready"); assert.equal(report.protocol, 3); -assert.deepEqual(report.features, ["executor_reuse", ...(["linux", "darwin", "win32"].includes(process.platform) ? ["workspace_directory", "local_runtime_v2", "workspace_functions", "workspace_structured_output", "workspace_tool_search", "workspace_mcp_http"] : []), "message_images", "function_result_images", "tool_search", "structured_output", "subagent_resources", "mcp_http_tools", "mcp_http_bearer_auth", "mcp_http_required", "workspace_tools", "workspace_prepare", "workspace_read", "workspace_command_observations"]); +assert.deepEqual(report.features, ["executor_reuse", ...(["linux", "darwin", "win32"].includes(process.platform) ? ["workspace_directory", "local_runtime_v2", "workspace_functions", "workspace_structured_output", "workspace_tool_search", "workspace_mcp_http"] : []), "message_images", "function_result_images", "tool_search", "structured_output", "subagent_resources", "mcp_http_tools", "mcp_http_bearer_auth", "mcp_http_required", "workspace_tools", "workspace_prepare", "workspace_command_observations"]); assert.equal(report.sdk, source.dependencies["@anthropic-ai/claude-agent-sdk"]); assert.equal(report.mcp, source.dependencies["@modelcontextprotocol/sdk"]); console.log(`Verified exported SDK ${report.sdk}, MCP ${report.mcp}, ${report.native}`); diff --git a/services/core/internal/runtimegateway/exchange_test.go b/services/core/internal/runtimegateway/exchange_test.go index 87d5db347..81a3fdd5d 100644 --- a/services/core/internal/runtimegateway/exchange_test.go +++ b/services/core/internal/runtimegateway/exchange_test.go @@ -16,7 +16,7 @@ func TestFramesMustNameTheRequestAssignment(t *testing.T) { defer s.Close("test") read := make(chan error, 1) go func() { - _, err := s.ReadWorkspaceFile(t.Context(), testAssignment, workspaceReadRequest()) + _, err := s.ListWorkspaceDirectory(t.Context(), testAssignment, workspaceReadRequest()) read <- err }() request := <-s.sendCh diff --git a/services/core/internal/runtimegateway/workspace_directory_test.go b/services/core/internal/runtimegateway/workspace_directory_test.go index aa104ea22..9821b7378 100644 --- a/services/core/internal/runtimegateway/workspace_directory_test.go +++ b/services/core/internal/runtimegateway/workspace_directory_test.go @@ -9,7 +9,7 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" ) -func TestWorkspaceDirectorySharesReadCorrelationAndFrameBound(t *testing.T) { +func TestWorkspaceDirectoryCorrelatesOneBoundedResult(t *testing.T) { s := NewSession(newFakeConn(), "device", "tenant", "test", nil, nil) defer s.Close("test") request := proto.WorkspaceReadPayload{Handle: "prepared", EnvironmentID: "environment", MaxEntries: proto.WorkspaceDirectoryMaxEntries} @@ -20,7 +20,7 @@ func TestWorkspaceDirectorySharesReadCorrelationAndFrameBound(t *testing.T) { }() message := <-s.sendCh var sent proto.WorkspaceReadPayload - if message.Type != proto.TypeWorkspaceRead || message.DecodePayload(&sent) != nil || sent.Operation != "directory" || sent.MaxBytes != 0 || sent.MaxEntries != request.MaxEntries { + if message.Type != proto.TypeWorkspaceRead || message.DecodePayload(&sent) != nil || sent != request { t.Fatal("directory request changed") } directory := &proto.WorkspaceDirectoryResult{Entries: make([]proto.WorkspaceDirectoryEntry, request.MaxEntries), Truncated: true} @@ -28,6 +28,8 @@ func TestWorkspaceDirectorySharesReadCorrelationAndFrameBound(t *testing.T) { directory.Entries[i] = proto.WorkspaceDirectoryEntry{Name: strings.Repeat("\x01", 250) + fmt.Sprintf("%04d", i), Kind: "directory"} } result := proto.WorkspaceReadResultPayload{Outcome: "completed", CloseAcknowledged: true, Directory: directory} + foreign, _ := proto.NewEnvelope(proto.TypeWorkspaceReadResult, "other-operation", result) + s.dispatch(foreign) reply, _ := message.Reply(proto.TypeWorkspaceReadResult, result) encoded, err := json.Marshal(reply) if err != nil || int64(len(encoded)) >= ReadLimit { @@ -37,10 +39,12 @@ func TestWorkspaceDirectorySharesReadCorrelationAndFrameBound(t *testing.T) { if err := <-done; err != nil { t.Fatal(err) } + if s.reg.LookupRun(message.ID) != nil { + t.Fatal("read registered a synthetic Run") + } } func TestWorkspaceDirectoryRejectsContradictoryAndUnboundedMetadata(t *testing.T) { - request := proto.WorkspaceReadPayload{Operation: "directory", MaxEntries: 2} size := int64(0) valid := proto.WorkspaceDirectoryEntry{Name: "file", Kind: "file", SizeBytes: &size} for _, entries := range [][]proto.WorkspaceDirectoryEntry{ @@ -49,17 +53,13 @@ func TestWorkspaceDirectoryRejectsContradictoryAndUnboundedMetadata(t *testing.T {{Name: strings.Repeat("x", 256), Kind: "directory"}}, } { result := proto.WorkspaceReadResultPayload{Outcome: "completed", CloseAcknowledged: true, Directory: &proto.WorkspaceDirectoryResult{Entries: entries}} - if validWorkspaceOperationResult(result, request) { + if validWorkspaceReadResult(result, 2) { t.Fatal("invalid directory metadata accepted") } } result := proto.WorkspaceReadResultPayload{Outcome: "completed", CloseAcknowledged: true, Directory: &proto.WorkspaceDirectoryResult{Entries: []proto.WorkspaceDirectoryEntry{valid}}} - if !validWorkspaceOperationResult(result, request) || validWorkspaceReadResult(result, 1024) { - t.Fatal("byte and directory result contracts mixed") - } - result.Data = []byte("unexpected bytes") - if validWorkspaceOperationResult(result, request) { - t.Fatal("contradictory result accepted") + if !validWorkspaceReadResult(result, 2) { + t.Fatal("valid directory metadata rejected") } } diff --git a/services/core/internal/runtimegateway/workspace_read.go b/services/core/internal/runtimegateway/workspace_read.go index cbb912f66..879c78c63 100644 --- a/services/core/internal/runtimegateway/workspace_read.go +++ b/services/core/internal/runtimegateway/workspace_read.go @@ -9,20 +9,8 @@ import ( "github.com/google/uuid" ) -// ReadWorkspaceFile observes one private operation; cancellation never retries or cancels native work. -func (s *Session) ReadWorkspaceFile(ctx context.Context, ref proto.AssignmentRef, request proto.WorkspaceReadPayload) (proto.WorkspaceReadResultPayload, error) { - if request.Operation != "" { - return proto.WorkspaceReadResultPayload{}, errors.New("agentdaemon gateway: invalid byte read operation") - } - return s.readWorkspace(ctx, ref, request) -} - +// ListWorkspaceDirectory observes one private operation; cancellation never retries or cancels native work. func (s *Session) ListWorkspaceDirectory(ctx context.Context, ref proto.AssignmentRef, request proto.WorkspaceReadPayload) (proto.WorkspaceReadResultPayload, error) { - request.Operation = "directory" - return s.readWorkspace(ctx, ref, request) -} - -func (s *Session) readWorkspace(ctx context.Context, ref proto.AssignmentRef, request proto.WorkspaceReadPayload) (proto.WorkspaceReadResultPayload, error) { var result proto.WorkspaceReadResultPayload if !proto.ValidWorkspaceReadRequest(request) { return result, errors.New("agentdaemon gateway: invalid workspace read") @@ -55,32 +43,17 @@ func (s *Session) readWorkspace(ctx context.Context, ref proto.AssignmentRef, re if err != nil { return result, err } - if reply.DecodePayload(&result) != nil || !validWorkspaceOperationResult(result, request) { + if reply.DecodePayload(&result) != nil || !validWorkspaceReadResult(result, request.MaxEntries) { return proto.WorkspaceReadResultPayload{}, errors.New("agentdaemon gateway: invalid workspace read response") } return result, nil } -func validWorkspaceOperationResult(result proto.WorkspaceReadResultPayload, request proto.WorkspaceReadPayload) bool { - if request.Operation == "directory" && result.Outcome == "completed" { - return result.CloseAcknowledged && result.ErrorCode == "" && len(result.Data) == 0 && !result.Truncated && proto.ValidWorkspaceDirectory(result.Directory, request.MaxEntries) - } - // Only a directory request can report that its path names no directory. - if request.Operation == "directory" && result.Outcome == "rejected" && result.ErrorCode == proto.WorkspaceReadNotDirectory { - return result.Directory == nil && len(result.Data) == 0 && !result.Truncated && !result.CloseAcknowledged - } - return validWorkspaceReadResult(result, request.MaxBytes) -} - func validWorkspaceReadResult(result proto.WorkspaceReadResultPayload, limit int) bool { - if result.Directory != nil { - return false - } if result.Outcome == "completed" { - return result.CloseAcknowledged && result.ErrorCode == "" && len(result.Data) <= limit && - (!result.Truncated || len(result.Data) == limit) + return result.CloseAcknowledged && result.ErrorCode == "" && proto.ValidWorkspaceDirectory(result.Directory, limit) } - if len(result.Data) != 0 || result.Truncated || result.CloseAcknowledged { + if result.Directory != nil || result.CloseAcknowledged { return false } if result.Outcome == "unknown" { @@ -90,7 +63,7 @@ func validWorkspaceReadResult(result proto.WorkspaceReadResultPayload, limit int return false } switch result.ErrorCode { - case "invalid_request", "resource_unavailable", "read_capacity", "read_unsupported", "not_found", "permission_denied", proto.AssignmentStale, proto.AssignmentConflict: + case "invalid_request", "resource_unavailable", "read_capacity", "read_unsupported", "not_found", "permission_denied", proto.WorkspaceReadNotDirectory, proto.AssignmentStale, proto.AssignmentConflict: return true default: return false diff --git a/services/core/internal/runtimegateway/workspace_read_test.go b/services/core/internal/runtimegateway/workspace_read_test.go index 6d3c89550..05a4e8026 100644 --- a/services/core/internal/runtimegateway/workspace_read_test.go +++ b/services/core/internal/runtimegateway/workspace_read_test.go @@ -1,9 +1,7 @@ package runtimegateway import ( - "bytes" "context" - "encoding/json" "errors" "strings" "testing" @@ -13,96 +11,35 @@ import ( ) func workspaceReadRequest() proto.WorkspaceReadPayload { - return proto.WorkspaceReadPayload{Handle: "prepared", EnvironmentID: "environment", Path: "file", MaxBytes: proto.WorkspaceReadMaxBytes} + return proto.WorkspaceReadPayload{Handle: "prepared", EnvironmentID: "environment", Path: "directory", MaxEntries: 2} } -func TestWorkspaceReadCorrelatesOneBoundedResult(t *testing.T) { - s := NewSession(newFakeConn(), "device", "tenant", "test", nil, nil) - defer s.Close("test") - done := make(chan error, 1) - data := bytes.Repeat([]byte{0, 127, 255, 3}, proto.WorkspaceReadMaxBytes/4) - go func() { - result, err := s.ReadWorkspaceFile(t.Context(), testAssignment, workspaceReadRequest()) - if err == nil && (!bytes.Equal(result.Data, data) || !result.Truncated || !result.CloseAcknowledged) { - err = errors.New("read data or acknowledgment differs") - } - done <- err - }() - request := <-s.sendCh - result := proto.WorkspaceReadResultPayload{Outcome: "completed", Data: data, Truncated: true, CloseAcknowledged: true} - foreign, _ := proto.NewEnvelope(proto.TypeWorkspaceReadResult, "other-operation", result) - s.dispatch(foreign) - reply, _ := request.Reply(proto.TypeWorkspaceReadResult, result) - encoded, err := json.Marshal(reply) - if err != nil || int64(len(encoded)) >= ReadLimit { - t.Fatal("result exceeds existing transport frame", err, len(encoded)) - } - s.dispatch(reply) - if err := <-done; err != nil { - t.Fatal(err) - } - if s.reg.LookupRun(request.ID) != nil { - t.Fatal("read registered a synthetic Run") - } -} - -func TestWorkspaceReadRejectsIncompleteOrContradictoryReplies(t *testing.T) { - for _, result := range []proto.WorkspaceReadResultPayload{ - {Outcome: "completed", Data: []byte("x")}, - {Outcome: "completed", CloseAcknowledged: true, Truncated: true}, - {Outcome: "completed", CloseAcknowledged: true, ErrorCode: "not_found"}, - {Outcome: "unknown", ErrorCode: "read_unconfirmed", Data: []byte("partial")}, - {Outcome: "rejected", ErrorCode: "native-private-secret"}, - } { - s := NewSession(newFakeConn(), "device", "tenant", "test", nil, nil) - done := make(chan error, 1) - go func() { - _, err := s.ReadWorkspaceFile(t.Context(), testAssignment, workspaceReadRequest()) - done <- err - }() - request := <-s.sendCh - reply, _ := request.Reply(proto.TypeWorkspaceReadResult, result) - s.dispatch(reply) - if err := <-done; err == nil { - t.Fatal("invalid result accepted", result.Outcome) - } - s.Close("test") - } -} - -func TestWorkspaceDirectoryAcceptsNotDirectoryOnlyForDirectoryReads(t *testing.T) { - directory := proto.WorkspaceReadPayload{Handle: "prepared", EnvironmentID: "environment", Path: "missing", MaxEntries: 2} +func TestWorkspaceReadValidatesRejections(t *testing.T) { for _, test := range []struct { - directory bool - result proto.WorkspaceReadResultPayload - accepted bool + result proto.WorkspaceReadResultPayload + accepted bool }{ - {true, proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: proto.WorkspaceReadNotDirectory}, true}, - {true, proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: "not_found"}, true}, - {true, proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: proto.WorkspaceReadNotDirectory, CloseAcknowledged: true}, false}, - {true, proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: proto.WorkspaceReadNotDirectory, Directory: &proto.WorkspaceDirectoryResult{Entries: []proto.WorkspaceDirectoryEntry{}}}, false}, - {true, proto.WorkspaceReadResultPayload{Outcome: "unknown", ErrorCode: proto.WorkspaceReadNotDirectory}, false}, - {true, proto.WorkspaceReadResultPayload{Outcome: "completed", CloseAcknowledged: true, ErrorCode: proto.WorkspaceReadNotDirectory, Directory: &proto.WorkspaceDirectoryResult{Entries: []proto.WorkspaceDirectoryEntry{}}}, false}, - {false, proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: proto.WorkspaceReadNotDirectory}, false}, - {false, proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: proto.AssignmentStale}, true}, - {true, proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: proto.AssignmentConflict}, true}, + {proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: proto.WorkspaceReadNotDirectory}, true}, + {proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: "not_found"}, true}, + {proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: proto.AssignmentStale}, true}, + {proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: proto.AssignmentConflict}, true}, + {proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: "native-private-secret"}, false}, + {proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: proto.WorkspaceReadNotDirectory, CloseAcknowledged: true}, false}, + {proto.WorkspaceReadResultPayload{Outcome: "rejected", ErrorCode: proto.WorkspaceReadNotDirectory, Directory: &proto.WorkspaceDirectoryResult{Entries: []proto.WorkspaceDirectoryEntry{}}}, false}, + {proto.WorkspaceReadResultPayload{Outcome: "unknown", ErrorCode: proto.WorkspaceReadNotDirectory}, false}, + {proto.WorkspaceReadResultPayload{Outcome: "completed", CloseAcknowledged: true, ErrorCode: proto.WorkspaceReadNotDirectory, Directory: &proto.WorkspaceDirectoryResult{Entries: []proto.WorkspaceDirectoryEntry{}}}, false}, } { s := NewSession(newFakeConn(), "device", "tenant", "test", nil, nil) done := make(chan error, 1) go func() { - var err error - if test.directory { - _, err = s.ListWorkspaceDirectory(t.Context(), testAssignment, directory) - } else { - _, err = s.ReadWorkspaceFile(t.Context(), testAssignment, workspaceReadRequest()) - } + _, err := s.ListWorkspaceDirectory(t.Context(), testAssignment, workspaceReadRequest()) done <- err }() request := <-s.sendCh reply, _ := request.Reply(proto.TypeWorkspaceReadResult, test.result) s.dispatch(reply) if err := <-done; (err == nil) != test.accepted { - t.Fatal("directory result validation changed", test.directory, test.result, err) + t.Fatal("directory result validation changed", test.result, err) } s.Close("test") } @@ -113,7 +50,7 @@ func TestWorkspaceReadObserverCancellationDoesNotSendCancelOrRetry(t *testing.T) defer s.Close("test") ctx, cancel := context.WithCancel(t.Context()) done := make(chan error, 1) - go func() { _, err := s.ReadWorkspaceFile(ctx, testAssignment, workspaceReadRequest()); done <- err }() + go func() { _, err := s.ListWorkspaceDirectory(ctx, testAssignment, workspaceReadRequest()); done <- err }() request := <-s.sendCh cancel() if err := <-done; !errors.Is(err, context.Canceled) { @@ -139,7 +76,7 @@ func TestWorkspaceReadCapacityAndConnectionLoss(t *testing.T) { done := make(chan error, 4) for i := 0; i < 4; i++ { go func() { - _, err := s.ReadWorkspaceFile(t.Context(), testAssignment, workspaceReadRequest()) + _, err := s.ListWorkspaceDirectory(t.Context(), testAssignment, workspaceReadRequest()) done <- err }() select { @@ -148,7 +85,7 @@ func TestWorkspaceReadCapacityAndConnectionLoss(t *testing.T) { t.Fatal("read not sent") } } - if _, err := s.ReadWorkspaceFile(t.Context(), testAssignment, workspaceReadRequest()); err == nil { + if _, err := s.ListWorkspaceDirectory(t.Context(), testAssignment, workspaceReadRequest()); err == nil { t.Fatal("capacity bypassed") } s.Close("connection lost") @@ -178,7 +115,7 @@ func TestWorkspaceReadRejectsOversizedRequestsBeforeQueueing(t *testing.T) { } ctx, cancel := context.WithTimeout(t.Context(), time.Second) defer cancel() - if _, err := s.ReadWorkspaceFile(ctx, testAssignment, request); err == nil || errors.Is(err, context.DeadlineExceeded) { + if _, err := s.ListWorkspaceDirectory(ctx, testAssignment, request); err == nil || errors.Is(err, context.DeadlineExceeded) { t.Fatal("oversized request not rejected before send", err) } select { diff --git a/services/core/tests/integration/environment_directory_test.go b/services/core/tests/integration/environment_directory_test.go index deb43aeda..0fc04e95b 100644 --- a/services/core/tests/integration/environment_directory_test.go +++ b/services/core/tests/integration/environment_directory_test.go @@ -80,7 +80,7 @@ func prepareDirectoryRead(t *testing.T, h *dispatchHarness, environment sessions h.write(frame.ID, proto.TypePreparationStatus, proto.PreparationStatusPayload{Handle: handle, Revision: 2, State: "ready"}) read := h.read(proto.TypeWorkspaceRead) var input proto.WorkspaceReadPayload - if read.DecodePayload(&input) != nil || input.Handle != handle || input.RunID != "" || input.EnvironmentID != environment.ID || input.Path != "reports" || input.Operation != "directory" { + if read.DecodePayload(&input) != nil || input.Handle != handle || input.RunID != "" || input.EnvironmentID != environment.ID || input.Path != "reports" { t.Fatal("directory request changed binding or path") } return frame.ID, read.ID