bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: hydrate issue context for worker spawns

This commit is contained in:
whoisasx 2026-07-03 23:00:49 +05:30
parent b3f59d6ad9
commit 7bd7729f4a
5 changed files with 281 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
backend/internal/daemon/lifecycle_wiring.go
View File

@ -111,11 +111,16 @@ func startSession(cfg config.Config, runtime runtimeselect.Runtime, store *sqlit
if err != nil {
logSCMProviderDisabled(log, err)
}
tracker, err := newGitHubTracker()
if err != nil {
logTrackerDisabled(log, err)
}
sessionSvc := sessionsvc.NewWithDeps(sessionsvc.Deps{
Manager: mgr,
Store: store,
PRClaimer: store,
SCM: scmProvider,
Tracker: tracker,
Telemetry: telemetry,
// no_signal only makes sense for harnesses whose adapters install
// activity hooks; the deriver registry is the source of truth for that.

21
backend/internal/daemon/tracker_wiring.go Normal file
View File

@ -0,0 +1,21 @@
package daemon
import (
"errors"
"log/slog"
trackergithub "github.com/aoagents/agent-orchestrator/backend/internal/adapters/tracker/github"
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
)
func newGitHubTracker() (ports.Tracker, error) {
return trackergithub.New(trackergithub.Options{Token: trackergithub.EnvTokenSource{EnvVars: []string{"AO_GITHUB_TOKEN"}}})
}
func logTrackerDisabled(logger *slog.Logger, err error) {
if errors.Is(err, trackergithub.ErrNoToken) {
logger.Warn("tracker issue prompt enrichment disabled: no usable GitHub token", "err", err)
} else {
logger.Warn("tracker issue prompt enrichment disabled: GitHub tracker setup failed", "err", err)
}
}

155
backend/internal/service/session/issue_context.go Normal file
View File

@ -0,0 +1,155 @@
package session
import (
"context"
"fmt"
"net/url"
"strconv"
"strings"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
)
const issueContextBodyLimit = 12000
func (s *Service) withIssueContext(ctx context.Context, cfg ports.SpawnConfig, project domain.ProjectRecord) ports.SpawnConfig {
if cfg.IssueContext != "" || cfg.IssueID == "" || s.tracker == nil {
return cfg
}
if cfg.Kind != "" && cfg.Kind != domain.KindWorker {
return cfg
}
id, ok := s.trackerIDForIssue(project, cfg.IssueID)
if !ok {
return cfg
}
issue, err := s.tracker.Get(ctx, id)
if err != nil {
return cfg
}
if issueContext := formatIssueContext(issue); issueContext != "" {
cfg.IssueContext = issueContext
}
return cfg
}
func (s *Service) trackerIDForIssue(project domain.ProjectRecord, issueID domain.IssueID) (domain.TrackerID, bool) {
issue := strings.TrimPrefix(strings.TrimSpace(string(issueID)), "#")
if issue == "" {
return domain.TrackerID{}, false
}
if native, ok := canonicalGitHubIssueNative(issue); ok {
return domain.TrackerID{Provider: domain.TrackerProviderGitHub, Native: native}, true
}
n, err := strconv.Atoi(issue)
if err != nil || n <= 0 {
return domain.TrackerID{}, false
}
repo, ok := s.githubRepoForTracker(project)
if !ok {
return domain.TrackerID{}, false
}
return domain.TrackerID{Provider: domain.TrackerProviderGitHub, Native: fmt.Sprintf("%s#%d", repo, n)}, true
}
func (s *Service) githubRepoForTracker(project domain.ProjectRecord) (string, bool) {
if s.scm != nil {
if repo, ok := s.scm.ParseRepository(project.RepoOriginURL); ok && repo.Provider == "github" && repo.Repo != "" {
return repo.Repo, true
}
}
owner, repo, err := githubRepoFromURL(project.RepoOriginURL)
if err != nil {
return "", false
}
return owner + "/" + repo, true
}
func canonicalGitHubIssueNative(raw string) (string, bool) {
if strings.Contains(raw, "://") {
return canonicalGitHubIssueURL(raw)
}
hash := strings.LastIndexByte(raw, '#')
if hash <= 0 || hash == len(raw)-1 {
return "", false
}
repo := strings.Trim(raw[:hash], "/")
owner, name, ok := splitIssueOwnerRepo(repo)
if !ok {
return "", false
}
n, err := strconv.Atoi(raw[hash+1:])
if err != nil || n <= 0 {
return "", false
}
return fmt.Sprintf("%s/%s#%d", owner, name, n), true
}
func splitIssueOwnerRepo(repo string) (string, string, bool) {
parts := strings.Split(strings.Trim(repo, "/"), "/")
if len(parts) != 2 {
return "", "", false
}
owner := strings.TrimSpace(parts[0])
name := strings.TrimSuffix(strings.TrimSpace(parts[1]), ".git")
return owner, name, owner != "" && name != ""
}
func canonicalGitHubIssueURL(raw string) (string, bool) {
u, err := url.Parse(raw)
if err != nil || !strings.EqualFold(u.Hostname(), "github.com") {
return "", false
}
parts := strings.Split(strings.Trim(u.Path, "/"), "/")
if len(parts) != 4 || parts[2] != "issues" {
return "", false
}
n, err := strconv.Atoi(parts[3])
if err != nil || n <= 0 {
return "", false
}
return fmt.Sprintf("%s/%s#%d", parts[0], strings.TrimSuffix(parts[1], ".git"), n), true
}
func formatIssueContext(issue domain.Issue) string {
var b strings.Builder
writeIssueLine(&b, "Issue", issue.ID.Native)
writeIssueLine(&b, "Title", issue.Title)
writeIssueLine(&b, "State", string(issue.State))
writeIssueLine(&b, "URL", issue.URL)
if len(issue.Labels) > 0 {
writeIssueLine(&b, "Labels", strings.Join(issue.Labels, ", "))
}
if len(issue.Assignees) > 0 {
writeIssueLine(&b, "Assignees", strings.Join(issue.Assignees, ", "))
}
body := strings.TrimSpace(domain.SanitizeControlChars(issue.Body))
if body != "" {
if b.Len() > 0 {
b.WriteString("\n\n")
}
b.WriteString("Body:\n")
b.WriteString(truncateIssueBody(body, issueContextBodyLimit))
}
return strings.TrimSpace(b.String())
}
func writeIssueLine(b *strings.Builder, label, value string) {
value = strings.TrimSpace(domain.SanitizeControlChars(value))
if value == "" {
return
}
if b.Len() > 0 {
b.WriteByte('\n')
}
fmt.Fprintf(b, "%s: %s", label, value)
}
func truncateIssueBody(body string, limit int) string {
runes := []rune(body)
if limit <= 0 || len(runes) <= limit {
return body
}
return string(runes[:limit]) + fmt.Sprintf("\n\n[Issue body truncated to %d characters.]", limit)
}

5
backend/internal/service/session/service.go
View File

@ -85,6 +85,7 @@ type Service struct {
store Store
prClaimer ports.PRClaimer
scm scmProvider
tracker ports.Tracker
clock func() time.Time
telemetry ports.EventSink
// signalCapable reports whether a harness has a hook pipeline that can
@ -107,6 +108,7 @@ type Deps struct {
Store Store
PRClaimer ports.PRClaimer
SCM scmProvider
Tracker ports.Tracker
Clock func() time.Time
Telemetry ports.EventSink
// SignalCapable gates the no_signal status downgrade per harness; daemon
@ -117,7 +119,7 @@ type Deps struct {
// NewWithDeps wires a session service with optional PR-claim dependencies.
func NewWithDeps(d Deps) *Service {
s := &Service{manager: d.Manager, store: d.Store, prClaimer: d.PRClaimer, scm: d.SCM, clock: d.Clock, signalCapable: d.SignalCapable, telemetry: d.Telemetry}
s := &Service{manager: d.Manager, store: d.Store, prClaimer: d.PRClaimer, scm: d.SCM, tracker: d.Tracker, clock: d.Clock, signalCapable: d.SignalCapable, telemetry: d.Telemetry}
if s.prClaimer == nil {
if w, ok := d.Store.(ports.PRClaimer); ok {
s.prClaimer = w
@ -140,6 +142,7 @@ func (s *Service) Spawn(ctx context.Context, cfg ports.SpawnConfig) (domain.Sess
if err != nil {
return domain.Session{}, fmt.Errorf("count sessions: %w", err)
}
cfg = s.withIssueContext(ctx, cfg, project)
rec, err := s.manager.Spawn(ctx, cfg)
if err != nil {
s.emitSpawnFailed(cfg, err, s.now().Sub(start).Milliseconds())

96
backend/internal/service/session/service_test.go
View File

@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"strings"
"testing"
"time"
@ -206,6 +207,7 @@ type fakeCommander struct {
cleanupErr error
spawnErr error
spawned bool
spawnedCfg ports.SpawnConfig
killsAtSpawn int
}
@ -214,6 +216,7 @@ func (f *fakeCommander) Spawn(_ context.Context, cfg ports.SpawnConfig) (domain.
return domain.SessionRecord{}, f.spawnErr
}
f.spawned = true
f.spawnedCfg = cfg
f.killsAtSpawn = len(f.killed)
return domain.SessionRecord{ID: "mer-9", ProjectID: cfg.ProjectID, Kind: cfg.Kind, Harness: cfg.Harness}, nil
}
@ -366,6 +369,99 @@ func TestSpawnEmitsFirstSessionOnboardingAndDuration(t *testing.T) {
}
}
type fakeTracker struct {
issue domain.Issue
err error
ids []domain.TrackerID
}
func (f *fakeTracker) Get(_ context.Context, id domain.TrackerID) (domain.Issue, error) {
f.ids = append(f.ids, id)
if f.err != nil {
return domain.Issue{}, f.err
}
return f.issue, nil
}
func (f *fakeTracker) List(context.Context, domain.TrackerRepo, domain.ListFilter) ([]domain.Issue, error) {
return nil, nil
}
func (f *fakeTracker) Preflight(context.Context) error { return nil }
func TestSpawnEnrichesIssueContextFromTracker(t *testing.T) {
st := newFakeStore()
st.projects["mer"] = domain.ProjectRecord{ID: "mer", RepoOriginURL: "https://github.com/acme/repo.git"}
fc := &fakeCommander{}
tracker := &fakeTracker{issue: domain.Issue{
ID: domain.TrackerID{Provider: domain.TrackerProviderGitHub, Native: "acme/repo#42"},
Title: "Fix generated prompts",
Body: "Prompt files should include standing instructions.",
State: domain.IssueInProgress,
URL: "https://github.com/acme/repo/issues/42",
Labels: []string{"bug", "prompts"},
Assignees: []string{"dev"},
}}
svc := NewWithDeps(Deps{Manager: fc, Store: st, Tracker: tracker})
if _, err := svc.Spawn(context.Background(), ports.SpawnConfig{ProjectID: "mer", Kind: domain.KindWorker, IssueID: "42"}); err != nil {
t.Fatalf("Spawn: %v", err)
}
if len(tracker.ids) != 1 || tracker.ids[0].Provider != domain.TrackerProviderGitHub || tracker.ids[0].Native != "acme/repo#42" {
t.Fatalf("tracker ids = %+v, want github acme/repo#42", tracker.ids)
}
issueContext := fc.spawnedCfg.IssueContext
for _, want := range []string{
"Issue: acme/repo#42",
"Title: Fix generated prompts",
"State: in_progress",
"URL: https://github.com/acme/repo/issues/42",
"Labels: bug, prompts",
"Assignees: dev",
"Body:\nPrompt files should include standing instructions.",
} {
if !strings.Contains(issueContext, want) {
t.Fatalf("IssueContext missing %q:\n%s", want, issueContext)
}
}
}
func TestSpawnIssueContextFetchFailureFallsBack(t *testing.T) {
st := newFakeStore()
st.projects["mer"] = domain.ProjectRecord{ID: "mer", RepoOriginURL: "https://github.com/acme/repo"}
fc := &fakeCommander{}
tracker := &fakeTracker{err: errors.New("tracker unavailable")}
svc := NewWithDeps(Deps{Manager: fc, Store: st, Tracker: tracker})
if _, err := svc.Spawn(context.Background(), ports.SpawnConfig{ProjectID: "mer", Kind: domain.KindWorker, IssueID: "42"}); err != nil {
t.Fatalf("Spawn: %v", err)
}
if len(tracker.ids) != 1 {
t.Fatalf("tracker calls = %d, want 1", len(tracker.ids))
}
if fc.spawnedCfg.IssueContext != "" {
t.Fatalf("IssueContext = %q, want fallback empty context", fc.spawnedCfg.IssueContext)
}
}
func TestSpawnIssueContextSkipsUnresolvableIssueRef(t *testing.T) {
st := newFakeStore()
st.projects["mer"] = domain.ProjectRecord{ID: "mer", RepoOriginURL: "https://github.com/acme/repo"}
fc := &fakeCommander{}
tracker := &fakeTracker{}
svc := NewWithDeps(Deps{Manager: fc, Store: st, Tracker: tracker})
if _, err := svc.Spawn(context.Background(), ports.SpawnConfig{ProjectID: "mer", Kind: domain.KindWorker, IssueID: "not-an-issue"}); err != nil {
t.Fatalf("Spawn: %v", err)
}
if len(tracker.ids) != 0 {
t.Fatalf("tracker calls = %d, want 0", len(tracker.ids))
}
if fc.spawnedCfg.IssueContext != "" {
t.Fatalf("IssueContext = %q, want empty", fc.spawnedCfg.IssueContext)
}
}
func TestSpawnFailedEmitsDuration(t *testing.T) {
st := newFakeStore()
st.projects["mer"] = domain.ProjectRecord{ID: "mer"}