GV

GeezerVison

The World Wide Web, restored to 800 pixels of glory.

Home|About|Gaming cabinet|Archives|How it works|Starter sites

On this site

  • Sign in
  • ActionsCI/CD and automation
  • CodespacesCloud development environments
  • All Features
  • Documentation
  • Training
  • Certifications
  • Build with Copilot
  • Customize Copilot
  • Integrations
  • Trending repos
  • GitHub Sponsors
  • Developer articles
  • Enterprises

This view

  • Original page
  • New address

GeezerVison is showing github.com — feat: create and run agent-local ACP sessions · coder/coder@d062169

File tree

Expand file treeCollapse file tree
Open diff view settings
Filter options
  • agent/x/agentacp
    • client.go
    • discovery.go
    • manager.go
    • manager_sessions.go
    • process.go
    • session.go
    • session_client.go
    • session_init.go
  • codersdk/workspacesdk
    • acp.go
Expand file treeCollapse file tree
Open diff view settings
Collapse file

‎agent/x/agentacp/client.go‎

Copy file name to clipboardExpand all lines: agent/x/agentacp/client.go
+10-1Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,16 @@ type discoveryClient struct{ client }
1414

1515
func (*discoveryClient) SessionUpdate(context.Context, acp.SessionNotification) error { return nil }
1616

17-
func (*client) RequestPermission(context.Context, acp.RequestPermissionRequest) (acp.RequestPermissionResponse, error) {
17+
func (*client) RequestPermission(ctx context.Context, req acp.RequestPermissionRequest) (acp.RequestPermissionResponse, error) {
18+
if ctx.Err() == nil {
19+
for _, kind := range []acp.PermissionOptionKind{acp.PermissionOptionKindAllowAlways, acp.PermissionOptionKindAllowOnce} {
20+
for _, option := range req.Options {
21+
if option.Kind == kind {
22+
return acp.RequestPermissionResponse{Outcome: acp.RequestPermissionOutcome{Selected: &acp.RequestPermissionOutcomeSelected{OptionId: option.OptionId}}}, nil
23+
}
24+
}
25+
}
26+
}
1827
//nolint:misspell // ACP names this protocol variant Cancelled.
1928
return acp.RequestPermissionResponse{Outcome: acp.RequestPermissionOutcome{Cancelled: &acp.RequestPermissionOutcomeCancelled{}}}, nil
2029
}
Collapse file

‎agent/x/agentacp/discovery.go‎

Copy file name to clipboardExpand all lines: agent/x/agentacp/discovery.go
+1Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -199,6 +199,7 @@ func (m *Manager) probe(ctx context.Context, cfg harnessConfig, directory string
199199
ctx, cancel := timeout(ctx, m.clock, setupTimeout)
200200
defer cancel()
201201
p, err := launch(ctx, cfg, directory, &discoveryClient{}, launchOptions{
202+
clock: m.clock,
202203
execer: m.execer,
203204
envInfo: m.envInfo,
204205
env: env,
Collapse file

‎agent/x/agentacp/manager.go‎

Copy file name to clipboardExpand all lines: agent/x/agentacp/manager.go
+78-5Lines changed: 78 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,12 @@
1-
// Package agentacp manages workspace-local ACP harnesses.
1+
// Package agentacp manages workspace-local ACP harnesses and sessions.
22
package agentacp
33

44
import (
55
"context"
6+
"maps"
67
"path/filepath"
78
"slices"
9+
"strings"
810
"sync"
911
"time"
1012

@@ -22,6 +24,14 @@ import (
2224
// ErrManagerClosed is returned when the manager has been closed.
2325
var ErrManagerClosed = xerrors.New("ACP manager closed")
2426

27+
// Errors classify agent API failures.
28+
var (
29+
ErrInvalid = xerrors.New("invalid ACP request")
30+
ErrNotFound = xerrors.New("ACP session not found")
31+
ErrConflict = xerrors.New("ACP request conflict")
32+
ErrUnavailable = xerrors.New("ACP harness unavailable")
33+
)
34+
2535
const setupTimeout = 30 * time.Second
2636

2737
// Options supplies the workspace execution environment and lifecycle.
@@ -33,12 +43,14 @@ type Options struct {
3343
EnvInfo usershell.EnvInfoer
3444
// UpdateEnv optionally enriches the inherited environment with workspace
3545
// variables and secrets. Discovery calls it once per reload and uses the
36-
// result for both cache invalidation and harness subprocesses.
46+
// result for both cache invalidation and harness subprocesses. Session
47+
// creation calls it again to use the current environment.
3748
UpdateEnv func([]string) ([]string, error)
3849
WorkingDir func() string
3950
}
4051

41-
// Manager owns discovery subprocesses and catalog snapshots.
52+
// Manager owns harness discovery and the session registry.
53+
// Each SessionFSM owns its process, tasks, operations, status, and transcript.
4254
type Manager struct {
4355
configHashes map[string][32]byte
4456
closedDone chan struct{}
@@ -53,6 +65,7 @@ type Manager struct {
5365
workingDir func() string
5466
mu sync.Mutex
5567
closed bool
68+
sessions map[workspacesdk.ACPSessionID]*SessionFSM
5669
configs map[string]harnessConfig
5770
configDir string
5871
discoveryDir string
@@ -66,7 +79,7 @@ type Manager struct {
6679

6780
// NewManager creates a manager; RunDiscovery starts filesystem discovery.
6881
// Clock, Execer, Filesystem, EnvInfo, and WorkingDir must be supplied.
69-
// UpdateEnv is optional and runs once per discovery reload.
82+
// UpdateEnv is optional and runs once per discovery reload or session launch.
7083
func NewManager(ctx context.Context, opts Options) *Manager {
7184
ctx, cancel := context.WithCancel(ctx)
7285
return &Manager{
@@ -79,6 +92,7 @@ func NewManager(ctx context.Context, opts Options) *Manager {
7992
envInfo: opts.EnvInfo,
8093
updateEnv: opts.UpdateEnv,
8194
workingDir: opts.WorkingDir,
95+
sessions: make(map[workspacesdk.ACPSessionID]*SessionFSM),
8296
closedDone: make(chan struct{}),
8397
configs: make(map[string]harnessConfig),
8498
configHashes: make(map[string][32]byte),
@@ -124,7 +138,51 @@ func (m *Manager) validate(directory string) error {
124138
return nil
125139
}
126140

127-
// Close stops discovery subprocesses and all manager-owned goroutines.
141+
// Create initializes a dedicated idle adapter.
142+
func (m *Manager) Create(ctx context.Context, req workspacesdk.ACPCreateSessionRequest) (workspacesdk.ACPSession, error) {
143+
s, err := m.createSession(ctx, req)
144+
if err != nil {
145+
return workspacesdk.ACPSession{}, err
146+
}
147+
r, err := s.Read(ctx, nil)
148+
if err != nil {
149+
m.discardSession(s)
150+
return workspacesdk.ACPSession{}, err
151+
}
152+
return r.Session, nil
153+
}
154+
155+
// Message submits a prompt to an existing idle session.
156+
func (m *Manager) Message(ctx context.Context, id workspacesdk.ACPSessionID, req workspacesdk.ACPMessageRequest) (workspacesdk.ACPMessageResponse, error) {
157+
if strings.TrimSpace(req.Text) == "" {
158+
return workspacesdk.ACPMessageResponse{}, ErrInvalid
159+
}
160+
s, err := m.get(ctx, id)
161+
if err != nil {
162+
return workspacesdk.ACPMessageResponse{}, err
163+
}
164+
return s.Message(ctx, req)
165+
}
166+
167+
// Interrupt obtains a session without recovering failures, then writes cancel.
168+
func (m *Manager) Interrupt(ctx context.Context, id workspacesdk.ACPSessionID) (workspacesdk.ACPSession, error) {
169+
s, err := m.get(ctx, id)
170+
if err != nil {
171+
return workspacesdk.ACPSession{}, err
172+
}
173+
return s.Interrupt(ctx)
174+
}
175+
176+
// Read preserves failed instances and their partial output.
177+
func (m *Manager) Read(ctx context.Context, id workspacesdk.ACPSessionID, after *workspacesdk.ACPCursor) (workspacesdk.ACPSessionResponse, error) {
178+
s, err := m.get(ctx, id)
179+
if err != nil {
180+
return workspacesdk.ACPSessionResponse{}, err
181+
}
182+
return s.Read(ctx, after)
183+
}
184+
185+
// Close prevents initialization and stops all session runtimes.
128186
func (m *Manager) Close() error {
129187
m.mu.Lock()
130188
if m.closed {
@@ -134,7 +192,11 @@ func (m *Manager) Close() error {
134192
}
135193
m.closed = true
136194
m.cancel()
195+
sessions := slices.Collect(maps.Values(m.sessions))
137196
m.mu.Unlock()
197+
for _, s := range sessions {
198+
s.Close()
199+
}
138200
m.wg.Wait()
139201
close(m.closedDone)
140202
return nil
@@ -205,6 +267,17 @@ func configOptions(options []acp.SessionConfigOption) []workspacesdk.ACPConfigOp
205267
return out
206268
}
207269

270+
func cloneEvent(e workspacesdk.ACPEvent) workspacesdk.ACPEvent {
271+
e.Update = slices.Clone(e.Update)
272+
if e.Session != nil {
273+
info := *e.Session
274+
info.Config = maps.Clone(info.Config)
275+
info.Harness = cloneHarness(info.Harness)
276+
e.Session = &info
277+
}
278+
return e
279+
}
280+
208281
func cloneHarness(h workspacesdk.ACPHarness) workspacesdk.ACPHarness {
209282
h.ConfigOptions = slices.Clone(h.ConfigOptions)
210283
for i := range h.ConfigOptions {
Collapse file

‎agent/x/agentacp/manager_sessions.go‎

Copy file name to clipboard
+139Lines changed: 139 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,139 @@
1+
package agentacp
2+
3+
import (
4+
"context"
5+
"maps"
6+
"path/filepath"
7+
8+
"golang.org/x/xerrors"
9+
10+
"github.com/coder/coder/v2/codersdk/workspacesdk"
11+
)
12+
13+
type initialization struct {
14+
done chan struct{}
15+
fsm *SessionFSM
16+
err error
17+
}
18+
19+
// createSession returns a new initialized FSM in idle.
20+
func (m *Manager) createSession(ctx context.Context, req workspacesdk.ACPCreateSessionRequest) (*SessionFSM, error) {
21+
if err := validateSessionRequest(m.fs, req.HarnessSlug, req.WorkingDirectory); err != nil {
22+
return nil, err
23+
}
24+
initReq := sessionInitRequest{
25+
harnessSlug: req.HarnessSlug,
26+
workingDirectory: req.WorkingDirectory,
27+
config: maps.Clone(req.Config),
28+
}
29+
if len(initReq.config) == 0 {
30+
initReq.config = nil
31+
}
32+
m.mu.Lock()
33+
if m.closed {
34+
m.mu.Unlock()
35+
return nil, ErrManagerClosed
36+
}
37+
attempt := &initialization{done: make(chan struct{})}
38+
m.wg.Go(func() {
39+
attempt.fsm, attempt.err = m.initializeAndRegisterSession(initReq)
40+
close(attempt.done)
41+
})
42+
m.mu.Unlock()
43+
return m.awaitInitialization(ctx, attempt)
44+
}
45+
46+
// initializeAndRegisterSession publishes only initialized FSMs. It runs in a
47+
// manager-owned goroutine so shutdown tracks in-flight setup.
48+
func (m *Manager) initializeAndRegisterSession(req sessionInitRequest) (*SessionFSM, error) {
49+
m.mu.Lock()
50+
cfg, found := m.configs[req.harnessSlug]
51+
m.mu.Unlock()
52+
if !found {
53+
return nil, xerrors.Errorf("%w: harness is missing or disabled", ErrUnavailable)
54+
}
55+
env := m.envInfo.Environ()
56+
if m.updateEnv != nil {
57+
var err error
58+
env, err = m.updateEnv(env)
59+
if err != nil {
60+
return nil, xerrors.Errorf("%w: prepare ACP environment: %v", ErrUnavailable, err)
61+
}
62+
}
63+
s, err := startSessionRuntime(m.ctx, cfg, req, sessionRuntimeOptions{
64+
launch: launchOptions{
65+
clock: m.clock,
66+
execer: m.execer,
67+
envInfo: m.envInfo,
68+
env: normalizeEnv(env),
69+
},
70+
fs: m.fs,
71+
logger: m.logger,
72+
})
73+
if err != nil {
74+
return nil, err
75+
}
76+
if err := m.registerSession(s); err != nil {
77+
s.Close()
78+
return nil, err
79+
}
80+
return s, nil
81+
}
82+
83+
func (m *Manager) registerSession(s *SessionFSM) error {
84+
m.mu.Lock()
85+
defer m.mu.Unlock()
86+
87+
if m.closed {
88+
return ErrManagerClosed
89+
}
90+
if m.sessions[s.id] != nil {
91+
return ErrConflict
92+
}
93+
m.sessions[s.id] = s
94+
return nil
95+
}
96+
97+
func (m *Manager) discardSession(s *SessionFSM) {
98+
// Keep the session registered until it closes so manager shutdown waits
99+
// for its process even when creation is canceled concurrently.
100+
s.Close()
101+
m.mu.Lock()
102+
if m.sessions[s.id] == s {
103+
delete(m.sessions, s.id)
104+
}
105+
m.mu.Unlock()
106+
}
107+
108+
func (m *Manager) awaitInitialization(ctx context.Context, attempt *initialization) (*SessionFSM, error) {
109+
select {
110+
case <-ctx.Done():
111+
<-attempt.done
112+
if attempt.fsm != nil {
113+
m.discardSession(attempt.fsm)
114+
}
115+
return nil, ctx.Err()
116+
case <-attempt.done:
117+
return attempt.fsm, attempt.err
118+
}
119+
}
120+
121+
// get returns an initialized session without creating or restoring one.
122+
func (m *Manager) get(ctx context.Context, id workspacesdk.ACPSessionID) (*SessionFSM, error) {
123+
if id.SessionID == "" || id.HarnessSlug == "" || !filepath.IsAbs(id.WorkingDirectory) {
124+
return nil, ErrInvalid
125+
}
126+
if err := ctx.Err(); err != nil {
127+
return nil, err
128+
}
129+
m.mu.Lock()
130+
defer m.mu.Unlock()
131+
if m.closed {
132+
return nil, ErrManagerClosed
133+
}
134+
s := m.sessions[id]
135+
if s == nil {
136+
return nil, ErrNotFound
137+
}
138+
return s, nil
139+
}

0 commit comments

Comments
0 (0)
You are visitor 01402520 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.