From 7bd7729f4ad8af6aa7df7dbbac5c5db71e6bcdd6 Mon Sep 17 00:00:00 2001 From: whoisasx Date: Fri, 3 Jul 2026 23:00:49 +0530 Subject: [PATCH] feat: hydrate issue context for worker spawns --- backend/internal/daemon/lifecycle_wiring.go | 5 + backend/internal/daemon/tracker_wiring.go | 21 +++ .../internal/service/session/issue_context.go | 155 ++++++++++++++++++ backend/internal/service/session/service.go | 5 +- .../internal/service/session/service_test.go | 96 +++++++++++ 5 files changed, 281 insertions(+), 1 deletion(-) create mode 100644 backend/internal/daemon/tracker_wiring.go create mode 100644 backend/internal/service/session/issue_context.go diff --git a/backend/internal/daemon/lifecycle_wiring.go b/backend/internal/daemon/lifecycle_wiring.go index c5391fe19..66994561f 100644 --- a/backend/internal/daemon/lifecycle_wiring.go +++ b/backend/internal/daemon/lifecycle_wiring.go @@ -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. diff --git a/backend/internal/daemon/tracker_wiring.go b/backend/internal/daemon/tracker_wiring.go new file mode 100644 index 000000000..a3b218451 --- /dev/null +++ b/backend/internal/daemon/tracker_wiring.go @@ -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) + } +} diff --git a/backend/internal/service/session/issue_context.go b/backend/internal/service/session/issue_context.go new file mode 100644 index 000000000..9067a8088 --- /dev/null +++ b/backend/internal/service/session/issue_context.go @@ -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) +} diff --git a/backend/internal/service/session/service.go b/backend/internal/service/session/service.go index 32d338897..92519814a 100644 --- a/backend/internal/service/session/service.go +++ b/backend/internal/service/session/service.go @@ -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()) diff --git a/backend/internal/service/session/service_test.go b/backend/internal/service/session/service_test.go index c5b1b1ca6..32313f6db 100644 --- a/backend/internal/service/session/service_test.go +++ b/backend/internal/service/session/service_test.go @@ -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"}