feat(agent): serve context sources over the agent socket (#26526)

## Overview

Split from #26466, scoped to **agent-only** changes. This PR exposes the
agent's context sources and snapshots over the existing agent socket.
There
are no changes outside `agent/`.

## What's included

- **agentsocket**: context source CRUD (`ContextSources`,
`GetContextSource`,
`AddContextSource`, `RemoveContextSource`) plus `GetContextSnapshot` and
`ResyncContext` RPCs, with matching client methods and proto. The server
receives the context `Manager` via `WithContextManager` and returns a
clean
  error when it is absent.
- **agentcontext**: the resync JSON response now carries the
per-resource
  `Name`, keeping the HTTP resync payload in sync with the drpc
  `PushContextState` path in `agentsocket`.
- **agent**: passes the context `Manager` to the socket server via
  `WithContextManager`.

## What's intentionally NOT here

- No MCP wiring. There are no MCP additions in `agent.go` or
`agentcontext`.
MCP ownership will land later in `agentcontext`; this PR does not build
on
  `agent/x/agentmcp`.
- No changes to `agent/x/agentmcp` or the `agentcontext` resolver. The
socket
  serves whatever context resources the `Manager` already resolves.

<details>
<summary>Context for reviewers</summary>

This is one of several PRs split out of #26466. Earlier revisions also
wired
live MCP servers through the socket; that scope was removed so this PR
stays
purely socket + context plumbing inside `agent/`. The agentcontext
resolver,
`agent/x/agentmcp`, and `agent.go` MCP startup behavior are unchanged
from
`main`.
</details>

---
_Created by Coder Agents on behalf of @kylecarbs._
This commit is contained in:
Kyle Carberry
2026-06-18 13:00:28 -07:00
committed by GitHub
parent 61fa2ab878
commit 992b1ffed1
9 changed files with 1968 additions and 83 deletions
+125 -1
View File
@@ -16,7 +16,8 @@ import (
type Option func(*options)
type options struct {
path string
path string
contextManager ContextManager
}
// WithPath sets the socket path. If not provided or empty, the client will
@@ -30,6 +31,14 @@ func WithPath(path string) Option {
}
}
// WithContextManager supplies the workspace-context Manager the server uses to
// serve context source CRUD. Server-only; ignored by the client.
func WithContextManager(cm ContextManager) Option {
return func(opts *options) {
opts.contextManager = cm
}
}
// Client provides a client for communicating with the workspace agentsocket API.
type Client struct {
client proto.DRPCAgentSocketClient
@@ -157,6 +166,92 @@ func (c *Client) UpdateAppStatus(ctx context.Context, req *agentproto.UpdateAppS
return c.client.UpdateAppStatus(ctx, req)
}
// ContextSources lists the workspace-context sources registered on the agent.
func (c *Client) ContextSources(ctx context.Context) ([]ContextSource, error) {
resp, err := c.client.ContextSources(ctx, &proto.ContextSourcesRequest{})
if err != nil {
return nil, err
}
sources := make([]ContextSource, 0, len(resp.Sources))
for _, s := range resp.Sources {
sources = append(sources, ContextSource{Path: s.GetPath()})
}
return sources, nil
}
// GetContextSource returns a single registered source. The path is
// canonicalized by the agent before matching.
func (c *Client) GetContextSource(ctx context.Context, path string) (ContextSource, error) {
resp, err := c.client.GetContextSource(ctx, &proto.GetContextSourceRequest{Path: path})
if err != nil {
return ContextSource{}, err
}
return ContextSource{Path: resp.GetSource().GetPath()}, nil
}
// AddContextSource registers a new scan root on the agent.
func (c *Client) AddContextSource(ctx context.Context, path string) (ContextSource, error) {
resp, err := c.client.AddContextSource(ctx, &proto.AddContextSourceRequest{Path: path})
if err != nil {
return ContextSource{}, err
}
return ContextSource{Path: resp.GetSource().GetPath()}, nil
}
// RemoveContextSource removes a previously-registered scan root.
func (c *Client) RemoveContextSource(ctx context.Context, path string) error {
_, err := c.client.RemoveContextSource(ctx, &proto.RemoveContextSourceRequest{Path: path})
return err
}
// GetContextSnapshot returns the agent's current resolved snapshot without
// forcing a re-walk.
func (c *Client) GetContextSnapshot(ctx context.Context) (ContextSnapshot, error) {
resp, err := c.client.GetContextSnapshot(ctx, &proto.ContextSnapshotRequest{})
if err != nil {
return ContextSnapshot{}, err
}
return contextSnapshotFromProto(resp.GetSnapshot()), nil
}
// ResyncContext forces a re-walk and synchronous push, returning the resulting
// snapshot. Use it as a barrier before fanning out a refresh.
func (c *Client) ResyncContext(ctx context.Context) (ContextSnapshot, error) {
resp, err := c.client.ResyncContext(ctx, &proto.ResyncContextRequest{})
if err != nil {
return ContextSnapshot{}, err
}
return contextSnapshotFromProto(resp.GetSnapshot()), nil
}
func contextSnapshotFromProto(s *proto.ContextSnapshot) ContextSnapshot {
if s == nil {
return ContextSnapshot{}
}
out := ContextSnapshot{
Version: s.GetVersion(),
AggregateHash: s.GetAggregateHash(),
Resources: make([]ContextResource, 0, len(s.GetResources())),
PayloadBytes: s.GetPayloadBytes(),
SnapshotError: s.GetSnapshotError(),
}
for _, r := range s.GetResources() {
out.Resources = append(out.Resources, ContextResource{
ID: r.GetId(),
Kind: r.GetKind(),
Source: r.GetSource(),
SourcePath: r.GetSourcePath(),
ContentHash: r.GetContentHash(),
SizeBytes: r.GetSizeBytes(),
Status: r.GetStatus(),
Error: r.GetError(),
Name: r.GetName(),
Description: r.GetDescription(),
})
}
return out
}
// SyncStatusResponse contains the status information for a unit.
type SyncStatusResponse struct {
UnitName unit.ID `table:"unit,default_sort" json:"unit_name"`
@@ -179,3 +274,32 @@ type DependencyInfo struct {
CurrentStatus unit.Status `table:"current status" json:"current_status"`
IsSatisfied bool `table:"satisfied" json:"is_satisfied"`
}
// ContextSource is a registered workspace-context scan root.
type ContextSource struct {
Path string `table:"path,default_sort" json:"path"`
}
// ContextResource is a resolved workspace-context resource. Payload bytes are
// never carried over the socket.
type ContextResource struct {
Kind string `table:"kind,default_sort" json:"kind"`
Name string `table:"name" json:"name"`
Source string `table:"source" json:"source"`
SourcePath string `table:"source path" json:"source_path"`
Status string `table:"status" json:"status"`
SizeBytes uint64 `table:"size bytes" json:"size_bytes"`
Error string `table:"error" json:"error"`
Description string `table:"-" json:"description"`
ID string `table:"-" json:"id"`
ContentHash string `table:"-" json:"content_hash"`
}
// ContextSnapshot is the agent's resolved workspace-context state.
type ContextSnapshot struct {
Version uint64 `json:"version"`
AggregateHash string `json:"aggregate_hash"`
Resources []ContextResource `json:"resources"`
PayloadBytes uint64 `json:"payload_bytes"`
SnapshotError string `json:"snapshot_error"`
}
+194
View File
@@ -0,0 +1,194 @@
package agentsocket_test
import (
"context"
"testing"
"github.com/stretchr/testify/require"
"cdr.dev/slog/v3"
"github.com/coder/coder/v2/agent/agentcontext"
"github.com/coder/coder/v2/agent/agentsocket"
"github.com/coder/coder/v2/testutil"
)
// fakeContextManager is an in-memory agentsocket.ContextManager for tests.
type fakeContextManager struct {
sources []agentcontext.Source
snapshot agentcontext.Snapshot
resyncErr error
resynced bool
}
func (f *fakeContextManager) Sources() []agentcontext.Source { return f.sources }
func (f *fakeContextManager) HasSource(path string) (string, bool) {
for _, s := range f.sources {
if s.Path == path {
return s.Path, true
}
}
return "", false
}
func (f *fakeContextManager) AddSource(s agentcontext.Source) (agentcontext.Source, error) {
for _, existing := range f.sources {
if existing.Path == s.Path {
return existing, nil
}
}
f.sources = append(f.sources, s)
return s, nil
}
func (f *fakeContextManager) RemoveSource(path string) error {
for i, s := range f.sources {
if s.Path == path {
f.sources = append(f.sources[:i], f.sources[i+1:]...)
return nil
}
}
return agentcontext.ErrSourceNotFound
}
func (f *fakeContextManager) Snapshot() agentcontext.Snapshot { return f.snapshot }
func (f *fakeContextManager) Resync(_ context.Context) (agentcontext.Snapshot, error) {
if f.resyncErr != nil {
return agentcontext.Snapshot{}, f.resyncErr
}
f.resynced = true
return f.snapshot, nil
}
func TestDRPCAgentSocketService_Context(t *testing.T) {
t.Parallel()
t.Run("SourceCRUDAndSnapshot", func(t *testing.T) {
t.Parallel()
const sourcePath = "/home/coder/project"
cm := &fakeContextManager{
snapshot: agentcontext.Snapshot{
Version: 7,
Resources: []agentcontext.Resource{{
ID: "instruction_file:" + sourcePath + "/AGENTS.md",
Kind: agentcontext.KindInstructionFile,
Source: sourcePath + "/AGENTS.md",
SourcePath: sourcePath,
SizeBytes: 42,
Status: agentcontext.StatusOK,
Description: "be concise",
}, {
// A built-in resource (no source path) the show filter must skip.
ID: "instruction_file:/home/coder/.coder/AGENTS.md",
Kind: agentcontext.KindInstructionFile,
Source: "/home/coder/.coder/AGENTS.md",
Status: agentcontext.StatusOK,
}},
},
}
socketPath := testutil.AgentSocketPath(t)
ctx := testutil.Context(t, testutil.WaitShort)
server, err := agentsocket.NewServer(
slog.Make().Leveled(slog.LevelDebug),
agentsocket.WithPath(socketPath),
agentsocket.WithContextManager(cm),
)
require.NoError(t, err)
defer server.Close()
client := newSocketClient(ctx, t, socketPath)
// Add a source.
src, err := client.AddContextSource(ctx, sourcePath)
require.NoError(t, err)
require.Equal(t, sourcePath, src.Path)
// It shows up in the list.
sources, err := client.ContextSources(ctx)
require.NoError(t, err)
require.Len(t, sources, 1)
require.Equal(t, sourcePath, sources[0].Path)
// Get the registered source.
got, err := client.GetContextSource(ctx, sourcePath)
require.NoError(t, err)
require.Equal(t, sourcePath, got.Path)
// Getting an unregistered source errors.
_, err = client.GetContextSource(ctx, "/nope")
require.Error(t, err)
// Snapshot carries resources with their source path stamped.
snap, err := client.GetContextSnapshot(ctx)
require.NoError(t, err)
require.EqualValues(t, 7, snap.Version)
require.Len(t, snap.Resources, 2)
require.Equal(t, agentcontext.KindInstructionFile.String(), snap.Resources[0].Kind)
require.Equal(t, sourcePath, snap.Resources[0].SourcePath)
require.EqualValues(t, 42, snap.Resources[0].SizeBytes)
// Remove the source; removing again reports not found.
require.NoError(t, client.RemoveContextSource(ctx, sourcePath))
err = client.RemoveContextSource(ctx, sourcePath)
require.Error(t, err)
require.Contains(t, err.Error(), "not found")
})
t.Run("Resync", func(t *testing.T) {
t.Parallel()
cm := &fakeContextManager{snapshot: agentcontext.Snapshot{Version: 3}}
socketPath := testutil.AgentSocketPath(t)
ctx := testutil.Context(t, testutil.WaitShort)
server, err := agentsocket.NewServer(
slog.Make().Leveled(slog.LevelDebug),
agentsocket.WithPath(socketPath),
agentsocket.WithContextManager(cm),
)
require.NoError(t, err)
defer server.Close()
client := newSocketClient(ctx, t, socketPath)
snap, err := client.ResyncContext(ctx)
require.NoError(t, err)
require.EqualValues(t, 3, snap.Version)
require.True(t, cm.resynced)
})
t.Run("NoManagerErrors", func(t *testing.T) {
t.Parallel()
socketPath := testutil.AgentSocketPath(t)
ctx := testutil.Context(t, testutil.WaitShort)
// No WithContextManager: the context RPCs must fail cleanly.
server, err := agentsocket.NewServer(
slog.Make().Leveled(slog.LevelDebug),
agentsocket.WithPath(socketPath),
)
require.NoError(t, err)
defer server.Close()
client := newSocketClient(ctx, t, socketPath)
// Every context RPC independently guards a nil context manager;
// exercise all of them so dropping a guard surfaces as a test
// failure rather than an agent-killing nil dereference in a DRPC
// handler.
_, err = client.ContextSources(ctx)
require.Error(t, err)
_, err = client.GetContextSource(ctx, "/tmp/x")
require.Error(t, err)
_, err = client.AddContextSource(ctx, "/tmp/x")
require.Error(t, err)
err = client.RemoveContextSource(ctx, "/tmp/x")
require.Error(t, err)
_, err = client.GetContextSnapshot(ctx)
require.Error(t, err)
_, err = client.ResyncContext(ctx)
require.Error(t, err)
})
}
File diff suppressed because it is too large Load Diff
+85
View File
@@ -70,6 +70,79 @@ message SyncListResponse {
repeated UnitInfo units = 1;
}
// ContextSource is a user-declared scan root the agent watches for
// workspace context (instruction files, skills, MCP configs) in
// addition to its built-in defaults. Identity is the canonical path.
message ContextSource {
string path = 1;
}
message ContextSourcesRequest {}
message ContextSourcesResponse {
repeated ContextSource sources = 1;
}
message GetContextSourceRequest {
string path = 1;
}
message GetContextSourceResponse {
ContextSource source = 1;
}
message AddContextSourceRequest {
string path = 1;
}
message AddContextSourceResponse {
ContextSource source = 1;
}
message RemoveContextSourceRequest {
string path = 1;
}
message RemoveContextSourceResponse {}
// ContextResource is the on-wire form of a resolved context resource.
// Payload bytes are never sent over the socket; they ship to coderd via
// the drpc PushContextState path. Mirrors agentcontext.Resource minus
// the payload and the tool list, which reach consumers via ListMCPTools.
message ContextResource {
string id = 1;
string kind = 2;
string source = 3;
string source_path = 4;
string content_hash = 5;
uint64 size_bytes = 6;
string status = 7;
string error = 8;
string name = 9;
string description = 10;
}
// ContextSnapshot is the agent's resolved context state.
message ContextSnapshot {
uint64 version = 1;
string aggregate_hash = 2;
repeated ContextResource resources = 3;
uint64 payload_bytes = 4;
string snapshot_error = 5;
}
message ContextSnapshotRequest {}
message ContextSnapshotResponse {
ContextSnapshot snapshot = 1;
}
message ResyncContextRequest {}
message ResyncContextResponse {
ContextSnapshot snapshot = 1;
}
// AgentSocket provides direct access to the agent over local IPC.
service AgentSocket {
// Ping the agent to check if it is alive.
@@ -88,4 +161,16 @@ service AgentSocket {
rpc SyncList(SyncListRequest) returns (SyncListResponse);
// Update app status, forwarded to coderd.
rpc UpdateAppStatus(coder.agent.v2.UpdateAppStatusRequest) returns (coder.agent.v2.UpdateAppStatusResponse);
// List the workspace context sources registered on the agent.
rpc ContextSources(ContextSourcesRequest) returns (ContextSourcesResponse);
// Get a single registered context source by path.
rpc GetContextSource(GetContextSourceRequest) returns (GetContextSourceResponse);
// Register a new context source (additional scan root).
rpc AddContextSource(AddContextSourceRequest) returns (AddContextSourceResponse);
// Remove a previously-registered context source.
rpc RemoveContextSource(RemoveContextSourceRequest) returns (RemoveContextSourceResponse);
// Return the agent's current resolved context snapshot without forcing a re-walk.
rpc GetContextSnapshot(ContextSnapshotRequest) returns (ContextSnapshotResponse);
// Force a re-walk and synchronous push, returning the resulting snapshot (barrier).
rpc ResyncContext(ResyncContextRequest) returns (ResyncContextResponse);
}
+241 -1
View File
@@ -47,6 +47,12 @@ type DRPCAgentSocketClient interface {
SyncStatus(ctx context.Context, in *SyncStatusRequest) (*SyncStatusResponse, error)
SyncList(ctx context.Context, in *SyncListRequest) (*SyncListResponse, error)
UpdateAppStatus(ctx context.Context, in *proto1.UpdateAppStatusRequest) (*proto1.UpdateAppStatusResponse, error)
ContextSources(ctx context.Context, in *ContextSourcesRequest) (*ContextSourcesResponse, error)
GetContextSource(ctx context.Context, in *GetContextSourceRequest) (*GetContextSourceResponse, error)
AddContextSource(ctx context.Context, in *AddContextSourceRequest) (*AddContextSourceResponse, error)
RemoveContextSource(ctx context.Context, in *RemoveContextSourceRequest) (*RemoveContextSourceResponse, error)
GetContextSnapshot(ctx context.Context, in *ContextSnapshotRequest) (*ContextSnapshotResponse, error)
ResyncContext(ctx context.Context, in *ResyncContextRequest) (*ResyncContextResponse, error)
}
type drpcAgentSocketClient struct {
@@ -131,6 +137,60 @@ func (c *drpcAgentSocketClient) UpdateAppStatus(ctx context.Context, in *proto1.
return out, nil
}
func (c *drpcAgentSocketClient) ContextSources(ctx context.Context, in *ContextSourcesRequest) (*ContextSourcesResponse, error) {
out := new(ContextSourcesResponse)
err := c.cc.Invoke(ctx, "/coder.agentsocket.v1.AgentSocket/ContextSources", drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcAgentSocketClient) GetContextSource(ctx context.Context, in *GetContextSourceRequest) (*GetContextSourceResponse, error) {
out := new(GetContextSourceResponse)
err := c.cc.Invoke(ctx, "/coder.agentsocket.v1.AgentSocket/GetContextSource", drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcAgentSocketClient) AddContextSource(ctx context.Context, in *AddContextSourceRequest) (*AddContextSourceResponse, error) {
out := new(AddContextSourceResponse)
err := c.cc.Invoke(ctx, "/coder.agentsocket.v1.AgentSocket/AddContextSource", drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcAgentSocketClient) RemoveContextSource(ctx context.Context, in *RemoveContextSourceRequest) (*RemoveContextSourceResponse, error) {
out := new(RemoveContextSourceResponse)
err := c.cc.Invoke(ctx, "/coder.agentsocket.v1.AgentSocket/RemoveContextSource", drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcAgentSocketClient) GetContextSnapshot(ctx context.Context, in *ContextSnapshotRequest) (*ContextSnapshotResponse, error) {
out := new(ContextSnapshotResponse)
err := c.cc.Invoke(ctx, "/coder.agentsocket.v1.AgentSocket/GetContextSnapshot", drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcAgentSocketClient) ResyncContext(ctx context.Context, in *ResyncContextRequest) (*ResyncContextResponse, error) {
out := new(ResyncContextResponse)
err := c.cc.Invoke(ctx, "/coder.agentsocket.v1.AgentSocket/ResyncContext", drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
type DRPCAgentSocketServer interface {
Ping(context.Context, *PingRequest) (*PingResponse, error)
SyncStart(context.Context, *SyncStartRequest) (*SyncStartResponse, error)
@@ -140,6 +200,12 @@ type DRPCAgentSocketServer interface {
SyncStatus(context.Context, *SyncStatusRequest) (*SyncStatusResponse, error)
SyncList(context.Context, *SyncListRequest) (*SyncListResponse, error)
UpdateAppStatus(context.Context, *proto1.UpdateAppStatusRequest) (*proto1.UpdateAppStatusResponse, error)
ContextSources(context.Context, *ContextSourcesRequest) (*ContextSourcesResponse, error)
GetContextSource(context.Context, *GetContextSourceRequest) (*GetContextSourceResponse, error)
AddContextSource(context.Context, *AddContextSourceRequest) (*AddContextSourceResponse, error)
RemoveContextSource(context.Context, *RemoveContextSourceRequest) (*RemoveContextSourceResponse, error)
GetContextSnapshot(context.Context, *ContextSnapshotRequest) (*ContextSnapshotResponse, error)
ResyncContext(context.Context, *ResyncContextRequest) (*ResyncContextResponse, error)
}
type DRPCAgentSocketUnimplementedServer struct{}
@@ -176,9 +242,33 @@ func (s *DRPCAgentSocketUnimplementedServer) UpdateAppStatus(context.Context, *p
return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
}
func (s *DRPCAgentSocketUnimplementedServer) ContextSources(context.Context, *ContextSourcesRequest) (*ContextSourcesResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
}
func (s *DRPCAgentSocketUnimplementedServer) GetContextSource(context.Context, *GetContextSourceRequest) (*GetContextSourceResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
}
func (s *DRPCAgentSocketUnimplementedServer) AddContextSource(context.Context, *AddContextSourceRequest) (*AddContextSourceResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
}
func (s *DRPCAgentSocketUnimplementedServer) RemoveContextSource(context.Context, *RemoveContextSourceRequest) (*RemoveContextSourceResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
}
func (s *DRPCAgentSocketUnimplementedServer) GetContextSnapshot(context.Context, *ContextSnapshotRequest) (*ContextSnapshotResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
}
func (s *DRPCAgentSocketUnimplementedServer) ResyncContext(context.Context, *ResyncContextRequest) (*ResyncContextResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
}
type DRPCAgentSocketDescription struct{}
func (DRPCAgentSocketDescription) NumMethods() int { return 8 }
func (DRPCAgentSocketDescription) NumMethods() int { return 14 }
func (DRPCAgentSocketDescription) Method(n int) (string, drpc.Encoding, drpc.Receiver, interface{}, bool) {
switch n {
@@ -254,6 +344,60 @@ func (DRPCAgentSocketDescription) Method(n int) (string, drpc.Encoding, drpc.Rec
in1.(*proto1.UpdateAppStatusRequest),
)
}, DRPCAgentSocketServer.UpdateAppStatus, true
case 8:
return "/coder.agentsocket.v1.AgentSocket/ContextSources", drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCAgentSocketServer).
ContextSources(
ctx,
in1.(*ContextSourcesRequest),
)
}, DRPCAgentSocketServer.ContextSources, true
case 9:
return "/coder.agentsocket.v1.AgentSocket/GetContextSource", drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCAgentSocketServer).
GetContextSource(
ctx,
in1.(*GetContextSourceRequest),
)
}, DRPCAgentSocketServer.GetContextSource, true
case 10:
return "/coder.agentsocket.v1.AgentSocket/AddContextSource", drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCAgentSocketServer).
AddContextSource(
ctx,
in1.(*AddContextSourceRequest),
)
}, DRPCAgentSocketServer.AddContextSource, true
case 11:
return "/coder.agentsocket.v1.AgentSocket/RemoveContextSource", drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCAgentSocketServer).
RemoveContextSource(
ctx,
in1.(*RemoveContextSourceRequest),
)
}, DRPCAgentSocketServer.RemoveContextSource, true
case 12:
return "/coder.agentsocket.v1.AgentSocket/GetContextSnapshot", drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCAgentSocketServer).
GetContextSnapshot(
ctx,
in1.(*ContextSnapshotRequest),
)
}, DRPCAgentSocketServer.GetContextSnapshot, true
case 13:
return "/coder.agentsocket.v1.AgentSocket/ResyncContext", drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCAgentSocketServer).
ResyncContext(
ctx,
in1.(*ResyncContextRequest),
)
}, DRPCAgentSocketServer.ResyncContext, true
default:
return "", nil, nil, nil, false
}
@@ -390,3 +534,99 @@ func (x *drpcAgentSocket_UpdateAppStatusStream) SendAndClose(m *proto1.UpdateApp
}
return x.CloseSend()
}
type DRPCAgentSocket_ContextSourcesStream interface {
drpc.Stream
SendAndClose(*ContextSourcesResponse) error
}
type drpcAgentSocket_ContextSourcesStream struct {
drpc.Stream
}
func (x *drpcAgentSocket_ContextSourcesStream) SendAndClose(m *ContextSourcesResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCAgentSocket_GetContextSourceStream interface {
drpc.Stream
SendAndClose(*GetContextSourceResponse) error
}
type drpcAgentSocket_GetContextSourceStream struct {
drpc.Stream
}
func (x *drpcAgentSocket_GetContextSourceStream) SendAndClose(m *GetContextSourceResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCAgentSocket_AddContextSourceStream interface {
drpc.Stream
SendAndClose(*AddContextSourceResponse) error
}
type drpcAgentSocket_AddContextSourceStream struct {
drpc.Stream
}
func (x *drpcAgentSocket_AddContextSourceStream) SendAndClose(m *AddContextSourceResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCAgentSocket_RemoveContextSourceStream interface {
drpc.Stream
SendAndClose(*RemoveContextSourceResponse) error
}
type drpcAgentSocket_RemoveContextSourceStream struct {
drpc.Stream
}
func (x *drpcAgentSocket_RemoveContextSourceStream) SendAndClose(m *RemoveContextSourceResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCAgentSocket_GetContextSnapshotStream interface {
drpc.Stream
SendAndClose(*ContextSnapshotResponse) error
}
type drpcAgentSocket_GetContextSnapshotStream struct {
drpc.Stream
}
func (x *drpcAgentSocket_GetContextSnapshotStream) SendAndClose(m *ContextSnapshotResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCAgentSocket_ResyncContextStream interface {
drpc.Stream
SendAndClose(*ResyncContextResponse) error
}
type drpcAgentSocket_ResyncContextStream struct {
drpc.Stream
}
func (x *drpcAgentSocket_ResyncContextStream) SendAndClose(m *ResyncContextResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_agent_agentsocket_proto_agentsocket_proto{}); err != nil {
return err
}
return x.CloseSend()
}
+3 -2
View File
@@ -44,8 +44,9 @@ func NewServer(logger slog.Logger, opts ...Option) (*Server, error) {
logger: logger,
path: options.path,
service: &DRPCAgentSocketService{
logger: logger,
unitManager: unit.NewManager(),
logger: logger,
unitManager: unit.NewManager(),
contextManager: options.contextManager,
},
}
+125 -4
View File
@@ -2,12 +2,14 @@ package agentsocket
import (
"context"
"encoding/hex"
"errors"
"sync"
"golang.org/x/xerrors"
"cdr.dev/slog/v3"
"github.com/coder/coder/v2/agent/agentcontext"
"github.com/coder/coder/v2/agent/agentsocket/proto"
agentproto "github.com/coder/coder/v2/agent/proto"
"github.com/coder/coder/v2/agent/unit"
@@ -16,14 +18,29 @@ import (
var _ proto.DRPCAgentSocketServer = (*DRPCAgentSocketService)(nil)
var (
ErrUnitManagerNotAvailable = xerrors.New("unit manager not available")
ErrAgentAPINotConnected = xerrors.New("agent not connected to coderd")
ErrUnitManagerNotAvailable = xerrors.New("unit manager not available")
ErrAgentAPINotConnected = xerrors.New("agent not connected to coderd")
ErrContextManagerNotAvailable = xerrors.New("context manager not available")
ErrContextSourceNotFound = xerrors.New("context source not found")
)
// ContextManager is the subset of *agentcontext.Manager the socket
// service needs to serve workspace-context source CRUD. It is an
// interface so tests can supply a fake.
type ContextManager interface {
Sources() []agentcontext.Source
HasSource(path string) (canonical string, ok bool)
AddSource(s agentcontext.Source) (agentcontext.Source, error)
RemoveSource(path string) error
Snapshot() agentcontext.Snapshot
Resync(ctx context.Context) (agentcontext.Snapshot, error)
}
// DRPCAgentSocketService implements the DRPC agent socket service.
type DRPCAgentSocketService struct {
unitManager *unit.Manager
logger slog.Logger
unitManager *unit.Manager
contextManager ContextManager
logger slog.Logger
mu sync.Mutex
agentAPI agentproto.DRPCAgentClient28
@@ -210,3 +227,107 @@ func (s *DRPCAgentSocketService) UpdateAppStatus(ctx context.Context, req *agent
}
return api.UpdateAppStatus(ctx, req)
}
// ContextSources lists the workspace-context sources registered on the agent.
func (s *DRPCAgentSocketService) ContextSources(_ context.Context, _ *proto.ContextSourcesRequest) (*proto.ContextSourcesResponse, error) {
if s.contextManager == nil {
return nil, ErrContextManagerNotAvailable
}
sources := s.contextManager.Sources()
out := &proto.ContextSourcesResponse{Sources: make([]*proto.ContextSource, 0, len(sources))}
for _, src := range sources {
out.Sources = append(out.Sources, &proto.ContextSource{Path: src.Path})
}
return out, nil
}
// GetContextSource returns a single registered source, canonicalizing the
// requested path before matching.
func (s *DRPCAgentSocketService) GetContextSource(_ context.Context, req *proto.GetContextSourceRequest) (*proto.GetContextSourceResponse, error) {
if s.contextManager == nil {
return nil, ErrContextManagerNotAvailable
}
canonical, ok := s.contextManager.HasSource(req.Path)
if !ok {
return nil, xerrors.Errorf("%q: %w", req.Path, ErrContextSourceNotFound)
}
return &proto.GetContextSourceResponse{Source: &proto.ContextSource{Path: canonical}}, nil
}
// AddContextSource registers a new scan root and triggers a re-resolve.
func (s *DRPCAgentSocketService) AddContextSource(_ context.Context, req *proto.AddContextSourceRequest) (*proto.AddContextSourceResponse, error) {
if s.contextManager == nil {
return nil, ErrContextManagerNotAvailable
}
src, err := s.contextManager.AddSource(agentcontext.Source{Path: req.Path})
if err != nil {
return nil, xerrors.Errorf("add context source: %w", err)
}
return &proto.AddContextSourceResponse{Source: &proto.ContextSource{Path: src.Path}}, nil
}
// RemoveContextSource removes a previously-registered scan root.
func (s *DRPCAgentSocketService) RemoveContextSource(_ context.Context, req *proto.RemoveContextSourceRequest) (*proto.RemoveContextSourceResponse, error) {
if s.contextManager == nil {
return nil, ErrContextManagerNotAvailable
}
if err := s.contextManager.RemoveSource(req.Path); err != nil {
if errors.Is(err, agentcontext.ErrSourceNotFound) {
return nil, xerrors.Errorf("%q: %w", req.Path, ErrContextSourceNotFound)
}
return nil, xerrors.Errorf("remove context source: %w", err)
}
return &proto.RemoveContextSourceResponse{}, nil
}
// GetContextSnapshot returns the agent's current resolved snapshot without
// forcing a re-walk.
func (s *DRPCAgentSocketService) GetContextSnapshot(_ context.Context, _ *proto.ContextSnapshotRequest) (*proto.ContextSnapshotResponse, error) {
if s.contextManager == nil {
return nil, ErrContextManagerNotAvailable
}
return &proto.ContextSnapshotResponse{Snapshot: contextSnapshotToProto(s.contextManager.Snapshot())}, nil
}
// ResyncContext forces a re-walk and synchronous push, returning the
// resulting snapshot. Callers use it as a barrier before fanning out a
// refresh.
func (s *DRPCAgentSocketService) ResyncContext(ctx context.Context, _ *proto.ResyncContextRequest) (*proto.ResyncContextResponse, error) {
if s.contextManager == nil {
return nil, ErrContextManagerNotAvailable
}
snap, err := s.contextManager.Resync(ctx)
if err != nil {
return nil, xerrors.Errorf("resync context: %w", err)
}
return &proto.ResyncContextResponse{Snapshot: contextSnapshotToProto(snap)}, nil
}
// contextSnapshotToProto converts an agentcontext.Snapshot to its on-wire
// form. Payload bytes are intentionally omitted; they reach coderd via the
// drpc PushContextState path. Keep the per-resource field mapping in sync
// with snapshotResponse in agent/agentcontext/api.go.
func contextSnapshotToProto(s agentcontext.Snapshot) *proto.ContextSnapshot {
out := &proto.ContextSnapshot{
Version: s.Version,
AggregateHash: hex.EncodeToString(s.AggregateHash[:]),
Resources: make([]*proto.ContextResource, 0, len(s.Resources)),
PayloadBytes: s.PayloadBytes,
SnapshotError: s.SnapshotError,
}
for _, r := range s.Resources {
out.Resources = append(out.Resources, &proto.ContextResource{
Id: r.ID,
Kind: r.Kind.String(),
Source: r.Source,
SourcePath: r.SourcePath,
ContentHash: hex.EncodeToString(r.ContentHash[:]),
SizeBytes: r.SizeBytes,
Status: r.Status.String(),
Error: r.Error,
Name: r.Name,
Description: r.Description,
})
}
return out
}