From 16904a97e8233e0030868cc393864ff294d8a957 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Tue, 14 Jul 2026 15:43:04 -0400 Subject: [PATCH 1/4] fix: reliability defaults, per-PR workspaces, honest budget messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four reliability fixes from live Windows testing, applied to BOTH ports (Python src/pr_af and the byte-compatible Go port) together: 1. Per-PR workspaces (parallel safety). _resolve_repo / ResolveRepo now key the checkout by repo AND PR number (-pr) when a PR number is known. The shared per-repo dir re-pointed its pr-review branch on every review, so two concurrent reviews of different PRs of the same repo silently reviewed the wrong checkout. Plain is kept when no PR number is known (repo_path/diff_text flows). New tests in both ports assert two PRs of the same repo resolve to different directories, each at its own PR head. 2. Realistic duration default. PR_AF_MAX_DURATION_SECONDS now defaults to 3600 (was 300); real reviews measure 60-70 minutes, so every fresh install died mid-pipeline. Changed in the resolve-caps cascade, input-schema and BudgetConfig defaults of both ports, both manifests, .env.example, both README env tables, and the docs. 3. Honest budget-exhaustion message. "Budget exhausted before " was reported whether the cost cap or the wall-clock cap tripped (in the Go port cost never accrues, so it was ALWAYS the wall clock). When the duration cap trips the message is now "Review time budget exceeded (max_duration_seconds=) before "; the cost cap keeps the historical wording. New §B.4 string is pinned byte- identically by tests in both ports. 4. Declare the harness idle watchdog. AGENTFIELD_HARNESS_IDLE_SECONDS (default "360") added to user_environment.optional in both manifests and both README env tables: harness CLIs in JSON mode emit events only at completion boundaries, so long single completions look silent — the old 120s SDK default killed healthy runs. Co-Authored-By: Claude Fable 5 --- .env.example | 5 +- README.md | 15 ++++++ agentfield-package.yaml | 8 ++- docs/ARCHITECTURE.md | 2 +- docs/DX.md | 2 +- go/README.md | 3 +- go/agentfield-package.yaml | 8 ++- go/internal/config/budget.go | 22 ++++---- go/internal/config/config_test.go | 15 +++--- go/internal/config/review.go | 8 +-- go/internal/orch/budget_test.go | 32 ++++++++++-- go/internal/orch/orchestrator.go | 29 +++++++++-- go/internal/orch/phases.go | 6 +-- go/internal/orch/resolve.go | 14 +++++- go/internal/orch/resolve_test.go | 84 +++++++++++++++++++++++++++++-- go/internal/schemas/input.go | 4 +- src/pr_af/app.py | 17 +++++-- src/pr_af/config.py | 2 +- src/pr_af/orchestrator.py | 23 +++++++-- src/pr_af/schemas/input.py | 2 +- tests/test_budget_env.py | 4 +- tests/test_budget_message.py | 39 ++++++++++++++ tests/test_resolve_repo.py | 72 ++++++++++++++++++++++---- 23 files changed, 347 insertions(+), 69 deletions(-) create mode 100644 tests/test_budget_message.py diff --git a/.env.example b/.env.example index 4fd3d1c..6832886 100644 --- a/.env.example +++ b/.env.example @@ -25,6 +25,9 @@ PR_AF_AI_INITIAL_BACKOFF_SECONDS=2.0 PR_AF_AI_MAX_BACKOFF_SECONDS=8.0 PR_AF_OPENCODE_BIN=opencode PR_AF_OPENCODE_SERVER= +# Harness no-output watchdog window (seconds). Harness CLIs in JSON mode emit +# events only at completion boundaries, so long single completions look silent. +AGENTFIELD_HARNESS_IDLE_SECONDS=360 # Alternate LLM provider keys (forwarded to harness when set) ANTHROPIC_API_KEY= @@ -33,7 +36,7 @@ GOOGLE_API_KEY= # --- Review budget / repo paths --- PR_AF_MAX_COST_USD=2.0 -PR_AF_MAX_DURATION_SECONDS=300 +PR_AF_MAX_DURATION_SECONDS=3600 PR_AF_WORKDIR=/workspaces # Defaults to cwd when unset PR_AF_REPO_PATH= diff --git a/README.md b/README.md index 9662b2b..6b2c35d 100644 --- a/README.md +++ b/README.md @@ -247,6 +247,21 @@ Poll for results: curl http://localhost:8080/api/v1/executions/ ``` +### Configuration (environment variables) + +The key knobs (see `.env.example` for the full list): + +| Variable | Purpose | +|-----------------------------|----------------------------------------------------------------| +| `OPENROUTER_API_KEY` | LLM provider key (OpenRouter) — required | +| `GH_TOKEN` | GitHub token (`repo` scope) for reading PRs and posting reviews | +| `PR_AF_PROVIDER` | Harness provider (default `opencode`) | +| `PR_AF_MODEL` | Harness model (default `openrouter/moonshotai/kimi-k2.5`) | +| `PR_AF_MAX_COST_USD` | Per-run cost ceiling in USD (default `2.0`) | +| `PR_AF_MAX_DURATION_SECONDS`| Per-run wall-clock ceiling in seconds (default `3600`) | +| `AGENTFIELD_HARNESS_IDLE_SECONDS` | Harness no-output watchdog window in seconds (default `360`) — harness CLIs in JSON mode emit events only at completion boundaries, so long single completions look silent | +| `PR_AF_WORKDIR` | Where PR checkouts live (default `/workspaces`); each PR gets its own `-pr` workspace | + ## GitHub Actions Integration The easiest way to use PR-AF is to drop it into your GitHub Actions. It requires **zero configuration** and runs securely using GitHub's built-in `GITHUB_TOKEN`. diff --git a/agentfield-package.yaml b/agentfield-package.yaml index a737174..8214dff 100644 --- a/agentfield-package.yaml +++ b/agentfield-package.yaml @@ -44,4 +44,10 @@ user_environment: default: "2.0" - name: PR_AF_MAX_DURATION_SECONDS description: Per-run wall-clock ceiling (seconds) - default: "300" + default: "3600" + # Harness CLIs in JSON mode emit events only at completion boundaries, so a + # long single completion looks silent; the old 120s SDK default killed + # healthy runs. + - name: AGENTFIELD_HARNESS_IDLE_SECONDS + description: harness no-output watchdog window (seconds) + default: "360" diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 3a359c1..047c6f3 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -549,7 +549,7 @@ class BudgetConfig(BaseModel): """All behavioral tuning in one place.""" # Global caps max_cost_usd: float = 2.0 - max_duration_seconds: int = 300 # 5 minutes + max_duration_seconds: int = 3600 # 60 minutes — real reviews measure 60-70 min # Phase-level max_concurrent_reviewers: int = 8 diff --git a/docs/DX.md b/docs/DX.md index 25271fe..1d64b4b 100644 --- a/docs/DX.md +++ b/docs/DX.md @@ -326,7 +326,7 @@ depth: auto # auto | quick | standard | deep # Budget budget: max_cost_usd: 2.00 - max_duration_seconds: 300 + max_duration_seconds: 3600 model_tier: standard # budget | standard | premium # Ignore patterns (glob) diff --git a/go/README.md b/go/README.md index dffc293..9ddbaf4 100644 --- a/go/README.md +++ b/go/README.md @@ -123,7 +123,8 @@ The node is configured entirely through the environment. | `PR_AF_PROVIDER` | Harness provider (default `opencode`) | | `PR_AF_MODEL` | Harness model (default `openrouter/moonshotai/kimi-k2.5`) | | `PR_AF_MAX_COST_USD` | Per-run cost ceiling in USD (default `2.0`) | -| `PR_AF_MAX_DURATION_SECONDS`| Per-run wall-clock ceiling in seconds (default `300`) | +| `PR_AF_MAX_DURATION_SECONDS`| Per-run wall-clock ceiling in seconds (default `3600`) | +| `AGENTFIELD_HARNESS_IDLE_SECONDS` | Harness no-output watchdog window in seconds (default `360`) — harness CLIs in JSON mode emit events only at completion boundaries, so long single completions look silent | | `HAX_API_KEY` | Optional — enables the HITL review-approval gate when set | Note: the code default model is `minimax/minimax-m2.5`, while the Docker image / diff --git a/go/agentfield-package.yaml b/go/agentfield-package.yaml index be24fab..31c4533 100644 --- a/go/agentfield-package.yaml +++ b/go/agentfield-package.yaml @@ -43,4 +43,10 @@ user_environment: default: "2.0" - name: PR_AF_MAX_DURATION_SECONDS description: per-run wall-clock ceiling - default: "300" + default: "3600" + # Harness CLIs in JSON mode emit events only at completion boundaries, so a + # long single completion looks silent; the old 120s SDK default killed + # healthy runs. + - name: AGENTFIELD_HARNESS_IDLE_SECONDS + description: harness no-output watchdog window (seconds) + default: "360" diff --git a/go/internal/config/budget.go b/go/internal/config/budget.go index 68776cd..df72107 100644 --- a/go/internal/config/budget.go +++ b/go/internal/config/budget.go @@ -4,8 +4,8 @@ import "strings" // BudgetConfig ports config.py BudgetConfig — global and per-phase budget caps. // Numeric defaults are the §D "Config numeric tables" verbatim. Note -// max_duration_seconds defaults to 1800 here but is always overridden per call -// to the resolved value (300 by default) via ReviewConfig.FromInput. +// max_duration_seconds is always overridden per call to the resolved value +// (3600 by default) via ReviewConfig.FromInput. type BudgetConfig struct { MaxCostUSD float64 `json:"max_cost_usd"` MaxDurationSeconds int `json:"max_duration_seconds"` @@ -49,7 +49,7 @@ func DefaultPhaseBudgets() map[string]float64 { func DefaultBudgetConfig() BudgetConfig { return BudgetConfig{ MaxCostUSD: 2.0, - MaxDurationSeconds: 1800, + MaxDurationSeconds: 3600, PhaseBudgets: DefaultPhaseBudgets(), MaxConcurrentReviewers: 8, MaxReferenceFollowsPerReviewer: 3, @@ -72,11 +72,13 @@ func evidencePackDefault() bool { } } -// ResolveBudgetCaps ports app.py:62-77 _resolve_budget_caps. An explicit -// argument always wins; when nil, the PR_AF_MAX_COST_USD / PR_AF_MAX_DURATION_ -// SECONDS env vars are consulted; finally the historical defaults 2.0 / 300. -// A malformed env value is an error (Python's float()/int() raises inside -// review(), where the ValueError class maps to HTTP 400). +// ResolveBudgetCaps ports app.py _resolve_budget_caps. An explicit argument +// always wins; when nil, the PR_AF_MAX_COST_USD / PR_AF_MAX_DURATION_SECONDS +// env vars are consulted; finally the defaults 2.0 / 3600 (real reviews +// measure 60-70 minutes — the historical 300s default killed every fresh +// install mid-pipeline). A malformed env value is an error (Python's +// float()/int() raises inside review(), where the ValueError class maps to +// HTTP 400). func ResolveBudgetCaps(maxCostUSD *float64, maxDurationSeconds *int) (float64, int, error) { cost := 2.0 if maxCostUSD != nil { @@ -88,12 +90,12 @@ func ResolveBudgetCaps(maxCostUSD *float64, maxDurationSeconds *int) (float64, i } } - dur := 300 + dur := 3600 if maxDurationSeconds != nil { dur = *maxDurationSeconds } else { var err error - if dur, err = intEnv("PR_AF_MAX_DURATION_SECONDS", 300); err != nil { + if dur, err = intEnv("PR_AF_MAX_DURATION_SECONDS", 3600); err != nil { return 0, 0, err } } diff --git a/go/internal/config/config_test.go b/go/internal/config/config_test.go index 7f26da4..d710378 100644 --- a/go/internal/config/config_test.go +++ b/go/internal/config/config_test.go @@ -75,7 +75,7 @@ func TestResolveBudgetCapsCascade(t *testing.T) { wantCost float64 wantDur int }{ - {"defaults when nil and no env", nil, nil, "", "", 2.0, 300}, + {"defaults when nil and no env", nil, nil, "", "", 2.0, 3600}, {"env used when arg nil", nil, nil, "3.3", "450", 3.3, 450}, {"explicit arg wins over env", ptrF(5.0), ptrI(120), "3.3", "450", 5.0, 120}, {"explicit arg wins over defaults", ptrF(1.25), ptrI(90), "", "", 1.25, 90}, @@ -245,8 +245,8 @@ func TestProviderEnv(t *testing.T) { func TestDefaultBudgetConfig(t *testing.T) { clearConfigEnv(t) b := DefaultBudgetConfig() - if b.MaxCostUSD != 2.0 || b.MaxDurationSeconds != 1800 { - t.Errorf("caps = %v/%d, want 2.0/1800", b.MaxCostUSD, b.MaxDurationSeconds) + if b.MaxCostUSD != 2.0 || b.MaxDurationSeconds != 3600 { + t.Errorf("caps = %v/%d, want 2.0/3600", b.MaxCostUSD, b.MaxDurationSeconds) } if b.MaxConcurrentReviewers != 8 || b.MaxReferenceFollowsPerReviewer != 3 || b.MaxChildSpawnsPerReviewer != 2 || b.MaxCrossRefDeepDives != 5 || @@ -397,15 +397,16 @@ func TestDefaultReviewConfigIgnorePaths(t *testing.T) { func TestFromInputBudgetCapsResolvedToPerCallDefault(t *testing.T) { // Key parity check: with no explicit caps and no env, FromInput must set the - // per-call duration to 300 (the resolved default) — NOT the 1800 BudgetConfig - // default, which Python always overwrites. + // per-call duration to 3600 via the ResolveBudgetCaps cascade — the same + // resolved value Python's review() always writes over the BudgetConfig + // default. clearConfigEnv(t) c := mustFromInput(t, schemas.ReviewInput{MaxReviewDepth: 2}) if c.Budget.MaxCostUSD != 2.0 { t.Errorf("MaxCostUSD = %v, want 2.0", c.Budget.MaxCostUSD) } - if c.Budget.MaxDurationSeconds != 300 { - t.Errorf("MaxDurationSeconds = %d, want 300 (per-call default, not 1800)", c.Budget.MaxDurationSeconds) + if c.Budget.MaxDurationSeconds != 3600 { + t.Errorf("MaxDurationSeconds = %d, want 3600 (per-call resolved default)", c.Budget.MaxDurationSeconds) } } diff --git a/go/internal/config/review.go b/go/internal/config/review.go index 9625f6e..d0db774 100644 --- a/go/internal/config/review.go +++ b/go/internal/config/review.go @@ -118,10 +118,10 @@ func DefaultReviewConfig() ReviewConfig { // // Budget caps: Python resolves max_cost_usd / max_duration_seconds in review() // BEFORE constructing the ReviewInput, so from_input always writes a concrete -// resolved value over the BudgetConfig defaults (which is why the 1800-second -// BudgetConfig default is dead — the per-call value is 300 by default). The Go -// ReviewInput carries these as *float64 / *int, so FromInput runs the same -// ResolveBudgetCaps cascade (explicit > env > 2.0/300) to reproduce that +// resolved value over the BudgetConfig defaults (the raw BudgetConfig duration +// default is dead — the per-call value is 3600 by default). The Go ReviewInput +// carries these as *float64 / *int, so FromInput runs the same +// ResolveBudgetCaps cascade (explicit > env > 2.0/3600) to reproduce that // effective value exactly. func (ReviewConfig) FromInput(in schemas.ReviewInput) (ReviewConfig, error) { c := DefaultReviewConfig() diff --git a/go/internal/orch/budget_test.go b/go/internal/orch/budget_test.go index d2e73cb..9ba0a39 100644 --- a/go/internal/orch/budget_test.go +++ b/go/internal/orch/budget_test.go @@ -32,8 +32,7 @@ func TestCostStaysInert(t *testing.T) { func TestWallClockBudgetTrips(t *testing.T) { // Production builds config via FromInput, which resolves the effective - // duration cap to 300s (ResolveBudgetCaps default) — not the 1800s raw - // BudgetConfig default that DefaultReviewConfig() carries. + // duration cap to 3600s (ResolveBudgetCaps default). cfg, cfgErr := config.ReviewConfig{}.FromInput(schemas.ReviewInput{}) if cfgErr != nil { t.Fatalf("FromInput: %v", cfgErr) @@ -46,13 +45,38 @@ func TestWallClockBudgetTrips(t *testing.T) { if o.isBudgetExhausted() { t.Error("budgetExhausted flag set prematurely") } - o.clock = func() time.Duration { return 400 * time.Second } + o.clock = func() time.Duration { return 4000 * time.Second } if !o.budgetOrTimeoutExhausted("intake") { - t.Error("should be exhausted past the 300s wall-clock cap") + t.Error("should be exhausted past the 3600s wall-clock cap") } if !o.isBudgetExhausted() { t.Error("budgetExhausted flag not set after timeout") } + // The wall-clock cap must be reported as a time budget, not a cost budget + // (§B.4 string — byte-identical to Python's _budget_exhausted_message). + want := "Review time budget exceeded (max_duration_seconds=3600) before intake" + if got := o.budgetExhaustedMessage("intake"); got != want { + t.Errorf("duration-cap message = %q, want %q", got, want) + } +} + +func TestCostCapKeepsBudgetExhaustedWording(t *testing.T) { + // When the COST cap trips (duration cap untouched), the historical + // "Budget exhausted before " wording is preserved (§B.4). + cfg, cfgErr := config.ReviewConfig{}.FromInput(schemas.ReviewInput{}) + if cfgErr != nil { + t.Fatalf("FromInput: %v", cfgErr) + } + o := New(Deps{App: &fakeApp{}}, schemas.ReviewInput{}, cfg) + o.clock = func() time.Duration { return 10 * time.Second } + o.registerCost("review", map[string]any{"cost_usd": cfg.Budget.MaxCostUSD}) + if !o.budgetOrTimeoutExhausted("review") { + t.Fatal("should be exhausted at the cost cap") + } + want := "Budget exhausted before review" + if got := o.budgetExhaustedMessage("review"); got != want { + t.Errorf("cost-cap message = %q, want %q", got, want) + } } func TestPhaseCapNeverTripsWithZeroCost(t *testing.T) { diff --git a/go/internal/orch/orchestrator.go b/go/internal/orch/orchestrator.go index d90a5f2..2b0887a 100644 --- a/go/internal/orch/orchestrator.go +++ b/go/internal/orch/orchestrator.go @@ -102,11 +102,12 @@ type Orchestrator struct { reviewID string startedAt time.Time - mu sync.Mutex // guards the counters below (mutated from fan-out goroutines) - totalCostUSD float64 - costBreakdown map[string]float64 - agentInvocations int - budgetExhausted bool + mu sync.Mutex // guards the counters below (mutated from fan-out goroutines) + totalCostUSD float64 + costBreakdown map[string]float64 + agentInvocations int + budgetExhausted bool + durationCapTripped bool // the wall-clock cap (not the cost cap) exhausted the budget // Single-threaded-written state (set before/after fan-outs, read after joins). prData *schemas.GitHubPRData @@ -426,12 +427,15 @@ func (o *Orchestrator) hitlMetadata(ctx context.Context) map[string]any { // budgetOrTimeoutExhausted ports _budget_or_timeout_exhausted. Cost stays 0 // (reasoner returns never carry cost_usd), so only the wall-clock check trips. +// Which cap tripped is recorded so budgetExhaustedMessage can word the failure +// honestly (§B.4 pair with Python's _budget_exhausted_message). func (o *Orchestrator) budgetOrTimeoutExhausted(phase string) bool { elapsed := o.clock().Seconds() o.mu.Lock() defer o.mu.Unlock() if elapsed > float64(o.config.Budget.MaxDurationSeconds) { o.budgetExhausted = true + o.durationCapTripped = true return true } if o.totalCostUSD >= o.config.Budget.MaxCostUSD { @@ -446,6 +450,21 @@ func (o *Orchestrator) budgetOrTimeoutExhausted(phase string) bool { return phaseSpent >= phaseCap } +// budgetExhaustedMessage words the exhaustion by cause: the wall-clock cap gets +// an explicit timeout message (in the Go port cost never accrues, so this is +// the only cap that actually trips), the cost cap keeps the historical +// "Budget exhausted before " wording. Byte-identical to Python's +// _budget_exhausted_message (§B.4). +func (o *Orchestrator) budgetExhaustedMessage(phase string) string { + o.mu.Lock() + defer o.mu.Unlock() + if o.durationCapTripped { + return fmt.Sprintf("Review time budget exceeded (max_duration_seconds=%d) before %s", + o.config.Budget.MaxDurationSeconds, phase) + } + return "Budget exhausted before " + phase +} + // registerCost ports _register_cost∘_extract_cost. extractCost reads "cost_usd" // off the reasoner return (always absent), so total stays 0.0. func (o *Orchestrator) registerCost(phase string, resultRaw map[string]any) { diff --git a/go/internal/orch/phases.go b/go/internal/orch/phases.go index 5879240..529c303 100644 --- a/go/internal/orch/phases.go +++ b/go/internal/orch/phases.go @@ -32,7 +32,7 @@ import ( func (o *Orchestrator) runIntake(ctx context.Context) (schemas.IntakeResult, error) { if o.budgetOrTimeoutExhausted("intake") { - return schemas.IntakeResult{}, budgetExhaustedErr("Budget exhausted before intake") + return schemas.IntakeResult{}, budgetExhaustedErr(o.budgetExhaustedMessage("intake")) } switch { @@ -90,7 +90,7 @@ func (o *Orchestrator) runIntake(ctx context.Context) (schemas.IntakeResult, err func (o *Orchestrator) runAnatomy(ctx context.Context, intake schemas.IntakeResult) (schemas.AnatomyResult, error) { if o.budgetOrTimeoutExhausted("anatomy") { - return schemas.AnatomyResult{}, budgetExhaustedErr("Budget exhausted before anatomy") + return schemas.AnatomyResult{}, budgetExhaustedErr(o.budgetExhaustedMessage("anatomy")) } if o.prData == nil { return schemas.AnatomyResult{}, errPRDataNotInitialized @@ -257,7 +257,7 @@ func (o *Orchestrator) runMetaSelectors( reviewDepth, reviewerFeedback string, ) (schemas.ReviewPlan, error) { if o.budgetOrTimeoutExhausted("meta_selectors") { - return schemas.ReviewPlan{}, budgetExhaustedErr("Budget exhausted before meta-selectors") + return schemas.ReviewPlan{}, budgetExhaustedErr(o.budgetExhaustedMessage("meta-selectors")) } lensFns := map[string]func(context.Context, reasoners.Deps, reasoners.MetaInput) (map[string]any, error){ diff --git a/go/internal/orch/resolve.go b/go/internal/orch/resolve.go index b68a419..c75d42f 100644 --- a/go/internal/orch/resolve.go +++ b/go/internal/orch/resolve.go @@ -97,6 +97,12 @@ func checkoutPRBranch(ctx context.Context, targetDir string, prNumber int) error // injected into a github.com HTTPS remote) and the PR branch checked out when a // PR number is known, and anything else falls back to $PR_AF_REPO_PATH / cwd. // +// The workspace is keyed by repo AND PR number when one is known +// (-pr): the shared per-repo dir re-pointed its pr-review branch +// on every review, so two concurrent reviews of different PRs of the same repo +// silently reviewed the wrong checkout. Plain is kept when no PR +// number is known (repo_path / diff_text flows), preserving the old layout. +// // Empty strings stand in for Python's None. Errors mirror Python's ValueError // ("git clone failed: …", plus the checkout strings via checkoutPRBranch). func ResolveRepo(ctx context.Context, repoPath, prURL string) (string, error) { @@ -135,7 +141,13 @@ func ResolveRepo(ctx context.Context, repoPath, prURL string) (string, error) { if target != "" && (strings.HasPrefix(target, "https://") || strings.HasPrefix(target, "http://") || strings.HasPrefix(target, "git@")) { repoName := strings.TrimSuffix(lastSegment(strings.TrimRight(target, "/")), ".git") - targetDir := filepath.Join(workdir, repoName) + workspaceName := repoName + if hasPR && prNumber != 0 { + // Per-PR workspace: isolates concurrent reviews of different PRs + // of the same repo (see the doc comment above). + workspaceName = fmt.Sprintf("%s-pr%d", repoName, prNumber) + } + targetDir := filepath.Join(workdir, workspaceName) if err := os.MkdirAll(workdir, 0o755); err != nil { return "", fmt.Errorf("git clone failed: %s", strings.TrimSpace(err.Error())) } diff --git a/go/internal/orch/resolve_test.go b/go/internal/orch/resolve_test.go index 325e450..940d63b 100644 --- a/go/internal/orch/resolve_test.go +++ b/go/internal/orch/resolve_test.go @@ -1,9 +1,11 @@ package orch -// V8: PR-branch checkout is idempotent — a reused workspace re-points the -// pr-review branch to the NEW PR head (the regression that previously reviewed -// every PR against the first one), and an unfetchable ref surfaces the exact -// "git fetch of PR #N head failed: …" error. Derived from tests/test_resolve_repo.py. +// V8: PR-branch checkout is idempotent — a reused workspace (same PR reviewed +// again) re-points the pr-review branch to the NEW head of that PR, and an +// unfetchable ref surfaces the exact "git fetch of PR #N head failed: …" error. +// ResolveRepo keys workspaces per PR (-pr) so concurrent reviews +// of different PRs of the same repo never share a checkout; plain +// is kept when no PR number is known. Derived from tests/test_resolve_repo.py. import ( "context" @@ -127,6 +129,80 @@ func TestCheckoutRaisesOnUnfetchablePRRef(t *testing.T) { } } +func TestResolveRepoKeysWorkspacePerPR(t *testing.T) { + // Two different PR numbers of the same repo must resolve to two different + // workspace directories, each checked out at its own PR head. + tmp := t.TempDir() + upstream := filepath.Join(tmp, "widgets") + makeUpstream(t, upstream) + publishPRRef(t, upstream, 1, "pr1") + publishPRRef(t, upstream, 2, "pr2") + + workdir := filepath.Join(tmp, "workspaces") + t.Setenv("PR_AF_WORKDIR", workdir) + if err := os.MkdirAll(workdir, 0o755); err != nil { + t.Fatal(err) + } + // Pre-seed the per-PR workspaces as clones of the local upstream so + // ResolveRepo takes the reuse path (fetch + PR checkout against the local + // origin) instead of cloning from github.com. autocrlf is disabled so the + // marker assertions below stay byte-exact on Windows checkouts. + for _, ws := range []string{"widgets-pr1", "widgets-pr2"} { + dir := filepath.Join(workdir, ws) + cloneWorkspace(t, upstream, dir) + git(t, dir, "config", "core.autocrlf", "false") + } + + dir1, err := ResolveRepo(context.Background(), "", "https://github.com/acme/widgets/pull/1") + if err != nil { + t.Fatalf("ResolveRepo(pull/1): %v", err) + } + dir2, err := ResolveRepo(context.Background(), "", "https://github.com/acme/widgets/pull/2") + if err != nil { + t.Fatalf("ResolveRepo(pull/2): %v", err) + } + + if dir1 == dir2 { + t.Fatalf("PR #1 and PR #2 share workspace %q — parallel reviews would collide", dir1) + } + if got := filepath.Base(dir1); got != "widgets-pr1" { + t.Errorf("PR #1 workspace = %q, want widgets-pr1", got) + } + if got := filepath.Base(dir2); got != "widgets-pr2" { + t.Errorf("PR #2 workspace = %q, want widgets-pr2", got) + } + if got := readFile(t, filepath.Join(dir1, "marker.txt")); got != "pr1\n" { + t.Errorf("PR #1 workspace marker = %q, want %q", got, "pr1\n") + } + if got := readFile(t, filepath.Join(dir2, "marker.txt")); got != "pr2\n" { + t.Errorf("PR #2 workspace marker = %q, want %q", got, "pr2\n") + } +} + +func TestResolveRepoPlainKeyWithoutPRNumber(t *testing.T) { + // repo_path/diff_text flows carry no PR number — the workspace stays keyed + // by plain , preserving the pre-per-PR layout. + tmp := t.TempDir() + upstream := filepath.Join(tmp, "widgets") + makeUpstream(t, upstream) + + workdir := filepath.Join(tmp, "workspaces") + t.Setenv("PR_AF_WORKDIR", workdir) + if err := os.MkdirAll(workdir, 0o755); err != nil { + t.Fatal(err) + } + // Pre-seed the plain-keyed workspace so no network clone is attempted. + git(t, tmp, "clone", "--depth", "1", "--no-tags", upstream, filepath.Join(workdir, "widgets")) + + dir, err := ResolveRepo(context.Background(), "https://github.com/acme/widgets.git", "") + if err != nil { + t.Fatalf("ResolveRepo(url, no PR): %v", err) + } + if got := filepath.Base(dir); got != "widgets" { + t.Errorf("workspace = %q, want plain widgets", got) + } +} + func TestExtractPRNumber(t *testing.T) { cases := []struct { url string diff --git a/go/internal/schemas/input.go b/go/internal/schemas/input.go index 2e47ba8..107a9c4 100644 --- a/go/internal/schemas/input.go +++ b/go/internal/schemas/input.go @@ -6,11 +6,11 @@ package schemas // // Divergence from schemas/input.py, deliberate (design §C.1): the pydantic // ReviewInput types max_cost_usd / max_duration_seconds as non-nullable float / -// int with concrete defaults (2.0 / 300). The Go struct instead mirrors the +// int with concrete defaults (2.0 / 3600). The Go struct instead mirrors the // actual node entry point — app.py review()'s signature, where both are // `... | None = None` — so they are *float64 / *int and stay nil until // config.ResolveBudgetCaps / config.ReviewConfig.FromInput resolve the -// explicit-arg > env > default (2.0 / 300) cascade. This keeps the "was a cap +// explicit-arg > env > default (2.0 / 3600) cascade. This keeps the "was a cap // explicitly supplied?" signal that Python recovers from the review() defaults. type ReviewInput struct { // Mode 1: GitHub PR URL. diff --git a/src/pr_af/app.py b/src/pr_af/app.py index fd5547b..66cfd6d 100644 --- a/src/pr_af/app.py +++ b/src/pr_af/app.py @@ -67,13 +67,15 @@ def _resolve_budget_caps( When the caller does not pass an explicit value, fall back to the ``PR_AF_MAX_COST_USD`` / ``PR_AF_MAX_DURATION_SECONDS`` env vars (so the budget can be tuned per deployment without a code change), and finally to - the historical defaults (2.0 USD / 300s) when neither is set. An explicit - argument always wins over the env var. + the defaults (2.0 USD / 3600s) when neither is set. An explicit argument + always wins over the env var. Real reviews measure 60-70 minutes, so the + duration default is a full hour — the historical 300s killed every fresh + install mid-pipeline. """ if max_cost_usd is None: max_cost_usd = float(os.getenv("PR_AF_MAX_COST_USD", "2.0")) if max_duration_seconds is None: - max_duration_seconds = int(os.getenv("PR_AF_MAX_DURATION_SECONDS", "300")) + max_duration_seconds = int(os.getenv("PR_AF_MAX_DURATION_SECONDS", "3600")) return max_cost_usd, max_duration_seconds @@ -125,7 +127,14 @@ def _resolve_repo(repo_path: str | None, pr_url: str | None) -> str: if isinstance(target, str) and target.startswith(("https://", "http://", "git@")): repo_name = target.rstrip("/").split("/")[-1].replace(".git", "") - target_dir = os.path.join(workdir, repo_name) + # Key the workspace by repo AND PR number when one is known + # (``-pr``): the shared per-repo dir re-pointed its + # ``pr-review`` branch on every review, so two concurrent reviews of + # different PRs of the same repo silently reviewed the wrong checkout. + # Plain ```` is kept when no PR number is known + # (repo_path / diff_text flows), preserving the old layout. + workspace_name = f"{repo_name}-pr{pr_number}" if pr_number else repo_name + target_dir = os.path.join(workdir, workspace_name) os.makedirs(workdir, exist_ok=True) clone_url = target diff --git a/src/pr_af/config.py b/src/pr_af/config.py index 65dc889..b923d45 100644 --- a/src/pr_af/config.py +++ b/src/pr_af/config.py @@ -23,7 +23,7 @@ class BudgetConfig(BaseModel): # Global caps max_cost_usd: float = 2.0 - max_duration_seconds: int = 1800 + max_duration_seconds: int = 3600 # Phase-level cost allocation (USD) phase_budgets: dict[str, float] = Field( diff --git a/src/pr_af/orchestrator.py b/src/pr_af/orchestrator.py index 89a5526..834b69b 100644 --- a/src/pr_af/orchestrator.py +++ b/src/pr_af/orchestrator.py @@ -117,6 +117,8 @@ def __init__(self, app: Any, input: ReviewInput, config: ReviewConfig | None = N self.cost_breakdown: dict[str, float] = {phase: 0.0 for phase in self.PHASE_ORDER} self.agent_invocations = 0 self.budget_exhausted = False + # True when the wall-clock cap (not the cost cap) exhausted the budget. + self.duration_cap_tripped = False self.meta_config = MetaSelectorConfig() self.pr_data: GitHubPRData | None = None @@ -388,7 +390,7 @@ def _hitl_metadata(self) -> dict[str, Any]: async def _run_intake(self) -> IntakeResult: if self._budget_or_timeout_exhausted("intake"): - raise BudgetExhaustedError("Budget exhausted before intake") + raise BudgetExhaustedError(self._budget_exhausted_message("intake")) if self.input.pr_url: client = GitHubClient() @@ -434,7 +436,7 @@ async def _run_intake(self) -> IntakeResult: async def _run_anatomy(self, intake: IntakeResult) -> AnatomyResult: if self._budget_or_timeout_exhausted("anatomy"): - raise BudgetExhaustedError("Budget exhausted before anatomy") + raise BudgetExhaustedError(self._budget_exhausted_message("anatomy")) if self.pr_data is None: raise RuntimeError("PR data not initialized") @@ -450,7 +452,7 @@ async def _run_anatomy(self, intake: IntakeResult) -> AnatomyResult: async def _run_planning(self, intake: IntakeResult, anatomy: AnatomyResult, review_depth: str) -> ReviewPlan: if self._budget_or_timeout_exhausted("planning"): - raise BudgetExhaustedError("Budget exhausted before planning") + raise BudgetExhaustedError(self._budget_exhausted_message("planning")) result_raw = await planning_phase( intake=intake.model_dump(), @@ -476,7 +478,7 @@ async def _run_meta_selectors( reviewer_feedback: str = "", ) -> ReviewPlan: if self._budget_or_timeout_exhausted("meta_selectors"): - raise BudgetExhaustedError("Budget exhausted before meta-selectors") + raise BudgetExhaustedError(self._budget_exhausted_message("meta-selectors")) lenses = self.meta_config.enabled_lenses lens_map = { @@ -1273,6 +1275,7 @@ def _budget_or_timeout_exhausted(self, phase: str) -> bool: elapsed = time.monotonic() - self.started_at if elapsed > self.config.budget.max_duration_seconds: self.budget_exhausted = True + self.duration_cap_tripped = True return True if self.total_cost_usd >= self.config.budget.max_cost_usd: self.budget_exhausted = True @@ -1281,6 +1284,18 @@ def _budget_or_timeout_exhausted(self, phase: str) -> bool: phase_cap = self.config.budget.phase_budgets.get(phase, float("inf")) return phase_spent >= phase_cap + def _budget_exhausted_message(self, phase: str) -> str: + """Word the exhaustion by cause: the wall-clock cap gets an explicit + timeout message, the cost cap keeps the historical "Budget exhausted + before " wording. Byte-identical to the Go port's + budgetExhaustedMessage (§B.4).""" + if self.duration_cap_tripped: + return ( + f"Review time budget exceeded " + f"(max_duration_seconds={self.config.budget.max_duration_seconds}) before {phase}" + ) + return f"Budget exhausted before {phase}" + def _register_cost(self, phase: str, cost: float | None) -> None: if cost is None: return diff --git a/src/pr_af/schemas/input.py b/src/pr_af/schemas/input.py index 4a798e3..7912a93 100644 --- a/src/pr_af/schemas/input.py +++ b/src/pr_af/schemas/input.py @@ -25,7 +25,7 @@ class ReviewInput(BaseModel): # Configuration overrides depth: str = "auto" # auto | quick | standard | deep max_cost_usd: float = 2.0 - max_duration_seconds: int = 300 + max_duration_seconds: int = 3600 focus: str = "auto" # auto | security | correctness | performance | tests ignore_paths: list[str] = Field(default_factory=list) hints: list[str] = Field(default_factory=list) # Project-specific review hints diff --git a/tests/test_budget_env.py b/tests/test_budget_env.py index 0259691..b38248b 100644 --- a/tests/test_budget_env.py +++ b/tests/test_budget_env.py @@ -3,7 +3,7 @@ Maps to the validation contract for ``_resolve_budget_caps``: * caller passes no value + env set -> env value is used -* caller passes no value + env unset -> historical defaults (2.0 USD / 300s) +* caller passes no value + env unset -> defaults (2.0 USD / 3600s) * caller passes an explicit value -> it wins over the env var """ @@ -29,7 +29,7 @@ def test_defaults_when_env_unset(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("PR_AF_MAX_DURATION_SECONDS", raising=False) monkeypatch.delenv("PR_AF_MAX_COST_USD", raising=False) cost, duration = _resolve_budget_caps(None, None) - assert duration == 300 + assert duration == 3600 assert cost == 2.0 diff --git a/tests/test_budget_message.py b/tests/test_budget_message.py new file mode 100644 index 0000000..dd51d77 --- /dev/null +++ b/tests/test_budget_message.py @@ -0,0 +1,39 @@ +"""Budget-exhaustion wording distinguishes the cause. + +The wall-clock cap gets an explicit "Review time budget exceeded +(max_duration_seconds=) before " message; the cost cap keeps the +historical "Budget exhausted before " wording. The strings are §B.4 +contracts — byte-identical to the Go port's ``budgetExhaustedMessage`` +(see go/internal/orch/budget_test.go). +""" + +from __future__ import annotations + +from pr_af.orchestrator import ReviewOrchestrator +from pr_af.schemas.input import ReviewInput + + +def _orchestrator() -> ReviewOrchestrator: + return ReviewOrchestrator(app=object(), input=ReviewInput(), config=None) + + +def test_duration_cap_message_names_the_time_budget() -> None: + orch = _orchestrator() + orch.config.budget.max_duration_seconds = 3600 + orch.started_at -= 4000 # simulate 4000s elapsed > 3600s cap + + assert orch._budget_or_timeout_exhausted("intake") + assert orch.duration_cap_tripped + assert ( + orch._budget_exhausted_message("intake") + == "Review time budget exceeded (max_duration_seconds=3600) before intake" + ) + + +def test_cost_cap_keeps_budget_exhausted_wording() -> None: + orch = _orchestrator() + orch.total_cost_usd = orch.config.budget.max_cost_usd + + assert orch._budget_or_timeout_exhausted("review") + assert not orch.duration_cap_tripped + assert orch._budget_exhausted_message("review") == "Budget exhausted before review" diff --git a/tests/test_resolve_repo.py b/tests/test_resolve_repo.py index 080f32a..6c4cc7b 100644 --- a/tests/test_resolve_repo.py +++ b/tests/test_resolve_repo.py @@ -1,13 +1,16 @@ -"""Behavior tests for PR-head checkout into the pr-af workspace. +"""Behavior tests for PR-head checkout and workspace keying. -Maps to the validation contract for ``_checkout_pr_branch``: +Maps to the validation contracts for ``_checkout_pr_branch`` and +``_resolve_repo``: * fresh workspace -> working tree reflects the PR head -* REUSED workspace (``pr-review`` already checked out from a prior PR) -> - working tree updates to the *new* PR head, not left on the prior PR's tree. - This is the regression test for the silent "no findings" bug: a reused - workspace kept reviewing every PR after the first against the first PR's tree. +* REUSED workspace (same PR reviewed again, ``pr-review`` already checked + out) -> working tree updates to the *new* head of that PR, not left on the + prior tree. This is the regression test for the silent "no findings" bug. * unfetchable PR ref -> raises, instead of silently leaving a stale tree. +* ``_resolve_repo`` keys workspaces per PR (``-pr``) so two + concurrent reviews of different PRs of the same repo never share a + checkout; plain ```` is kept when no PR number is known. Uses real temporary git repositories (a local path acts as the ``origin`` remote with ``refs/pull//head`` refs); no network and no mocking of the @@ -17,14 +20,11 @@ from __future__ import annotations import subprocess -from typing import TYPE_CHECKING +from pathlib import Path import pytest -from pr_af.app import _checkout_pr_branch - -if TYPE_CHECKING: - from pathlib import Path +from pr_af.app import _checkout_pr_branch, _resolve_repo def _git(*args: str, cwd: Path) -> str: @@ -108,3 +108,53 @@ def test_checkout_raises_on_unfetchable_pr_ref(tmp_path: Path) -> None: with pytest.raises(ValueError, match="PR #999"): _checkout_pr_branch(str(target), 999) + + +def test_resolve_repo_keys_workspace_per_pr( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Two different PR numbers of the same repo resolve to different dirs.""" + upstream = tmp_path / "widgets" + _make_upstream(upstream) + _publish_pr_ref(upstream, 1, "pr1") + _publish_pr_ref(upstream, 2, "pr2") + + workdir = tmp_path / "workspaces" + workdir.mkdir() + monkeypatch.setenv("PR_AF_WORKDIR", str(workdir)) + # Pre-seed the per-PR workspaces as clones of the local upstream so + # _resolve_repo takes the reuse path (fetch + PR checkout against the + # local origin) instead of cloning from github.com. autocrlf is disabled + # so the marker assertions below stay byte-exact on Windows checkouts. + for workspace in ("widgets-pr1", "widgets-pr2"): + _clone_workspace(upstream, workdir / workspace) + _git("config", "core.autocrlf", "false", cwd=workdir / workspace) + + dir1 = _resolve_repo(None, "https://github.com/acme/widgets/pull/1") + dir2 = _resolve_repo(None, "https://github.com/acme/widgets/pull/2") + + assert dir1 != dir2, "parallel reviews of different PRs would collide" + assert Path(dir1).name == "widgets-pr1" + assert Path(dir2).name == "widgets-pr2" + assert (Path(dir1) / "marker.txt").read_text() == "pr1\n" + assert (Path(dir2) / "marker.txt").read_text() == "pr2\n" + + +def test_resolve_repo_plain_key_without_pr_number( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """repo_path/diff_text flows (no PR number) keep the plain-name workspace.""" + upstream = tmp_path / "widgets" + _make_upstream(upstream) + + workdir = tmp_path / "workspaces" + workdir.mkdir() + monkeypatch.setenv("PR_AF_WORKDIR", str(workdir)) + # Pre-seed the plain-keyed workspace so no network clone is attempted. + _git( + "clone", "--depth", "1", "--no-tags", + str(upstream), str(workdir / "widgets"), cwd=tmp_path, + ) + + resolved = _resolve_repo("https://github.com/acme/widgets.git", None) + assert Path(resolved).name == "widgets" From ea1562d9f2da89c501924b73b30643f0b8e73494 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Tue, 14 Jul 2026 15:43:16 -0400 Subject: [PATCH 2/4] chore(deps): bump AgentField Go SDK to 2cc5fe2a MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Picks up fix/harness-windows-reliability — the Windows stdin fix for opencode prompts (v0.0.0-20260714191100-2cc5fe2adcf4). Co-Authored-By: Claude Fable 5 --- go/go.mod | 2 +- go/go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go/go.mod b/go/go.mod index 3516eb0..339de0a 100644 --- a/go/go.mod +++ b/go/go.mod @@ -5,7 +5,7 @@ module github.com/Agent-Field/pr-af/go go 1.21 require ( - github.com/Agent-Field/agentfield/sdk/go v0.0.0-20260713163335-795bdd57b4de + github.com/Agent-Field/agentfield/sdk/go v0.0.0-20260714191100-2cc5fe2adcf4 github.com/golang-jwt/jwt/v5 v5.3.1 github.com/invopop/jsonschema v0.13.0 github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 diff --git a/go/go.sum b/go/go.sum index 5f16fd9..92df3d9 100644 --- a/go/go.sum +++ b/go/go.sum @@ -1,5 +1,5 @@ -github.com/Agent-Field/agentfield/sdk/go v0.0.0-20260713163335-795bdd57b4de h1:GYwOpQQheZ53FgmPta3PJ2CKQjATnPIMwAvNsBA7vF4= -github.com/Agent-Field/agentfield/sdk/go v0.0.0-20260713163335-795bdd57b4de/go.mod h1:08VZk14uw4GJH6a34psHkuLu+DcRr197Zi0IGmLlfrM= +github.com/Agent-Field/agentfield/sdk/go v0.0.0-20260714191100-2cc5fe2adcf4 h1:B3uCMLZSa2rsRDRGuecBIXCgkYqBuScSK4CXKPDaCgk= +github.com/Agent-Field/agentfield/sdk/go v0.0.0-20260714191100-2cc5fe2adcf4/go.mod h1:08VZk14uw4GJH6a34psHkuLu+DcRr197Zi0IGmLlfrM= github.com/bahlo/generic-list-go v0.2.0 h1:5sz/EEAK+ls5wF+NeqDpk5+iNdMDXrh3z3nPnH1Wvgk= github.com/bahlo/generic-list-go v0.2.0/go.mod h1:2KvAjgMlE5NNynlg/5iLrrCCZ2+5xWbdbCW3pNTGyYg= github.com/buger/jsonparser v1.1.1 h1:2PnMjfWD7wBILjqQbt530v576A/cAbQvEW9gGIpYMUs= From 77e017c00c4c7a56730e48a658b9c2b6a4621b65 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Tue, 14 Jul 2026 16:10:23 -0400 Subject: [PATCH 3/4] feat(go): route pipeline phases through CallLocal to restore the review DAG MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Python node reports every pipeline phase (intake_phase, anatomy_phase, meta_* selectors, each review_dimension, adversary_phase, ...) as a tracked child execution of the parent review — the control-plane UI renders the run as that orchestration DAG. The Go port called the phase functions directly in-process, so the control plane only ever saw a single review execution: the architecture's DAG collapsed into one opaque node. Route the orchestrator's reasoner seams through the SDK's Agent.CallLocal when orch.Deps.Local is wired (production always wires the live agent): each phase invocation now builds a child execution context and emits running/succeeded/failed workflow events under its registered reasoner name, which the control plane mirrors into the executions table — one DAG node per phase, parented under review, same shape as Python. - reasoners: add Name* constants as the single source of truth for the registered names; register.go and the CallLocal seams share them so DAG phase names cannot drift from the registered surface. - afx: add ToMap (Bind's inverse). It keeps nested field values typed instead of JSON-round-tripping the whole struct — a plain round trip flattens OrderedPatches into a Go map and silently re-sorts diff_patches, breaking the insertion order the meta/obligation prompts render. - orch: LocalCaller interface + Deps.Local; callLocalSeams/viaLocal adapt each typed seam to CallLocal under its registered name. nil Local keeps the direct-call seams (unit tests, stub harnesses). - node: BuildAgent wires localCaller to the live agent; reviewHandler forwards it as Deps.Local. Behavior is otherwise unchanged: a CallLocal-routed phase runs the same reasoners.* function with the same deps via the registered handler, events are fire-and-forget (an unreachable control plane never fails a review), and the ToMap->Bind round trip is covered by tests including deliberate zero values on fields with non-zero bind-side defaults. Co-Authored-By: Claude Fable 5 --- go/internal/afx/bind.go | 66 ++++++++ go/internal/afx/bind_test.go | 83 ++++++++++ go/internal/node/calllocal_wiring_test.go | 61 ++++++++ go/internal/node/node.go | 7 + go/internal/node/register.go | 38 ++--- go/internal/orch/calllocal_test.go | 178 ++++++++++++++++++++++ go/internal/orch/orchestrator.go | 80 ++++++++++ go/internal/reasoners/names.go | 25 +++ 8 files changed, 521 insertions(+), 17 deletions(-) create mode 100644 go/internal/node/calllocal_wiring_test.go create mode 100644 go/internal/orch/calllocal_test.go create mode 100644 go/internal/reasoners/names.go diff --git a/go/internal/afx/bind.go b/go/internal/afx/bind.go index 104650a..9b00224 100644 --- a/go/internal/afx/bind.go +++ b/go/internal/afx/bind.go @@ -5,6 +5,8 @@ package afx import ( "encoding/json" "fmt" + "reflect" + "strings" ) // Bind decodes a reasoner's untyped input map into a typed value T. @@ -33,3 +35,67 @@ func Bind[T any](input map[string]any) (T, error) { } return out, nil } + +// ToMap is Bind's inverse: it renders a typed input struct as the +// map[string]any shape the SDK's reasoner handlers (and Agent.CallLocal) +// accept. Top-level exported fields become map entries keyed by their json +// tag, and the field VALUES stay typed — deliberately NOT a marshal→unmarshal +// round trip, which would decode nested values into plain Go maps and lose +// what their custom marshalers encode. The concrete casualty would be +// reasoners.OrderedPatches: its object-order MarshalJSON preserves Python's +// dict insertion order, but flattened into a map[string]any it re-marshals +// with sorted keys and the meta/obligation prompts would render patches in +// the wrong order. Keeping values typed lets Bind on the handler side (and +// the SDK's workflow-event emitter) re-marshal them through the same custom +// marshalers, so ToMap→Bind is lossless. +// +// The reasoner input structs are flat, fully json-tagged, and carry no +// omitempty (every key is emitted, so Bind-side default seeding never +// overrides a deliberately zero field); ToMap ignores omitempty accordingly. +// Anonymous embedded structs without their own json tag are flattened the way +// encoding/json flattens them. +func ToMap(v any) (map[string]any, error) { + rv := reflect.ValueOf(v) + for rv.Kind() == reflect.Pointer { + if rv.IsNil() { + return nil, fmt.Errorf("afx.ToMap: nil %T", v) + } + rv = rv.Elem() + } + if rv.Kind() != reflect.Struct { + return nil, fmt.Errorf("afx.ToMap: %T is not a struct", v) + } + out := make(map[string]any, rv.NumField()) + fillMap(out, rv) + return out, nil +} + +// fillMap writes rv's fields into out, recursing through untagged anonymous +// struct fields (encoding/json flattening). +func fillMap(out map[string]any, rv reflect.Value) { + rt := rv.Type() + for i := 0; i < rt.NumField(); i++ { + f := rt.Field(i) + if !f.IsExported() { + continue + } + name, _, _ := strings.Cut(f.Tag.Get("json"), ",") + if name == "-" { + continue + } + if name == "" { + if f.Anonymous { + fv := rv.Field(i) + for fv.Kind() == reflect.Pointer && !fv.IsNil() { + fv = fv.Elem() + } + if fv.Kind() == reflect.Struct { + fillMap(out, fv) + continue + } + } + name = f.Name + } + out[name] = rv.Field(i).Interface() + } +} diff --git a/go/internal/afx/bind_test.go b/go/internal/afx/bind_test.go index 8027a9c..c5aebca 100644 --- a/go/internal/afx/bind_test.go +++ b/go/internal/afx/bind_test.go @@ -163,3 +163,86 @@ func TestBind_TypeMismatch(t *testing.T) { t.Fatalf("Bind = nil error, want unmarshal error for string into int") } } + +// orderedVal is a marshaler-carrying type standing in for +// reasoners.OrderedPatches: MarshalJSON controls its bytes. ToMap must keep +// values typed so this marshaler still runs when the map is re-marshaled — +// the property that preserves diff_patches insertion order through the +// CallLocal round trip. +type orderedVal struct{ order []string } + +func (o orderedVal) MarshalJSON() ([]byte, error) { + out := "{" + for i, k := range o.order { + if i > 0 { + out += "," + } + b, _ := json.Marshal(k) + out += string(b) + ":1" + } + return []byte(out + "}"), nil +} + +// TestToMap_KeepsValuesTyped proves ToMap does not flatten field values into +// plain maps: the custom marshaler's byte order survives a re-marshal of the +// produced map. +func TestToMap_KeepsValuesTyped(t *testing.T) { + type in struct { + Patches orderedVal `json:"diff_patches"` + Depth string `json:"depth"` + } + m, err := ToMap(in{Patches: orderedVal{order: []string{"z.go", "a.go"}}, Depth: "quick"}) + if err != nil { + t.Fatalf("ToMap: %v", err) + } + if m["depth"] != "quick" { + t.Errorf("depth = %v, want quick", m["depth"]) + } + b, err := json.Marshal(m["diff_patches"]) + if err != nil { + t.Fatalf("re-marshal diff_patches: %v", err) + } + if string(b) != `{"z.go":1,"a.go":1}` { + t.Errorf("diff_patches bytes = %s, want insertion order preserved", b) + } +} + +// TestToMap_TagHandling covers tag name selection, json:"-" skipping, +// unexported skipping, untagged-field fallback to the Go name, and anonymous +// embedded flattening. +func TestToMap_TagHandling(t *testing.T) { + type Embedded struct { + Inner string `json:"inner"` + } + type in struct { + Embedded + Named string `json:"named_key"` + Skipped string `json:"-"` + Bare string + hidden string //nolint:unused // proves unexported fields are skipped + } + m, err := ToMap(&in{Embedded: Embedded{Inner: "i"}, Named: "n", Skipped: "s", Bare: "b"}) + if err != nil { + t.Fatalf("ToMap: %v", err) + } + want := map[string]any{"inner": "i", "named_key": "n", "Bare": "b"} + if len(m) != len(want) { + t.Fatalf("map = %#v, want %#v", m, want) + } + for k, v := range want { + if m[k] != v { + t.Errorf("m[%q] = %v, want %v", k, m[k], v) + } + } +} + +// TestToMap_Errors covers the non-struct and nil-pointer error paths. +func TestToMap_Errors(t *testing.T) { + if _, err := ToMap([]string{"not", "a", "struct"}); err == nil { + t.Error("ToMap(slice) = nil error, want non-struct error") + } + var p *struct{} + if _, err := ToMap(p); err == nil { + t.Error("ToMap(nil pointer) = nil error, want nil error message") + } +} diff --git a/go/internal/node/calllocal_wiring_test.go b/go/internal/node/calllocal_wiring_test.go new file mode 100644 index 0000000..aea2907 --- /dev/null +++ b/go/internal/node/calllocal_wiring_test.go @@ -0,0 +1,61 @@ +package node + +// DAG wiring: the review handler must hand the node's tracked local-call seam +// to the orchestrator (orch.Deps.Local), and BuildAgent must point that seam +// at the live SDK agent — that is the link that turns the in-process pipeline +// phases into control-plane child executions (the review DAG). + +import ( + "context" + "testing" + + "github.com/Agent-Field/pr-af/go/internal/config" + "github.com/Agent-Field/pr-af/go/internal/orch" + "github.com/Agent-Field/pr-af/go/internal/schemas" +) + +// recordingLocal is a sentinel orch.LocalCaller. +type recordingLocal struct{} + +func (recordingLocal) CallLocal(context.Context, string, map[string]any) (any, error) { + return map[string]any{}, nil +} + +func TestReviewHandlerForwardsLocalCaller(t *testing.T) { + repo := t.TempDir() // existing dir -> ResolveRepo returns it, no clone/network. + sentinel := recordingLocal{} + + var got orch.Deps + n := &Node{ + NodeID: "pr-af-go", + reviewApp: &fakeApp{}, + localCaller: sentinel, + runReview: func(_ context.Context, deps orch.Deps, _ schemas.ReviewInput, _ config.ReviewConfig) (schemas.ReviewResult, error) { + got = deps + return schemas.ReviewResult{}, nil + }, + } + + if _, err := n.reviewHandler(context.Background(), map[string]any{"repo_path": repo}); err != nil { + t.Fatalf("reviewHandler: %v", err) + } + if got.Local != orch.LocalCaller(sentinel) { + t.Fatalf("orch.Deps.Local = %#v, want the node's localCaller seam", got.Local) + } +} + +func TestBuildAgentWiresLocalCallerToApp(t *testing.T) { + t.Setenv("NODE_ID", "") + t.Setenv("AGENTFIELD_SERVER", "") + t.Setenv("AGENTFIELD_API_KEY", "") + t.Setenv("PORT", "") + t.Setenv("OPENROUTER_API_KEY", "") + + n, err := BuildAgent("pr-af-go", "8007", "desc") + if err != nil { + t.Fatalf("BuildAgent: %v", err) + } + if n.localCaller != orch.LocalCaller(n.App) { + t.Fatal("BuildAgent must wire localCaller to the live agent (App)") + } +} diff --git a/go/internal/node/node.go b/go/internal/node/node.go index 7043489..c8ab815 100644 --- a/go/internal/node/node.go +++ b/go/internal/node/node.go @@ -60,6 +60,12 @@ type Node struct { // runReview leave it unused (nil). gh github.Client + // localCaller is the tracked same-process invocation seam fed into + // orch.Deps.Local: production points it at App so every pipeline phase is + // reported to the control plane as a child execution (the review DAG). + // Tests that stub runReview or need direct-call phases leave it nil. + localCaller orch.LocalCaller + // runReview is the orchestrator-construct-and-run seam. Production builds and // runs the real orchestrator; the error-mapping tests inject ErrBadInput / // other failures without a live harness (design §F "seam for the orchestrator @@ -170,6 +176,7 @@ func BuildAgent(defaultNodeID, defaultPort, description string) (*Node, error) { ListenAddress: ":" + port, reviewApp: app, gh: github.NewClient(""), // reads GH_TOKEN internally (app.py GitHubClient()) + localCaller: app, runReview: defaultRunReview, tags: map[string][]string{}, } diff --git a/go/internal/node/register.go b/go/internal/node/register.go index a07731a..6e6ff18 100644 --- a/go/internal/node/register.go +++ b/go/internal/node/register.go @@ -39,23 +39,26 @@ func (n *Node) RegisterAll() { // The 16 router reasoners, tagged ["review","pr"], in §B.1 order. Each is // bound (afx.Bind) into its typed input and backed by a reasoners.Deps built - // from the agent (Harness=App, AI=App). - regReasoner(n, "intake_phase", reasoners.IntakePhase) - regReasoner(n, "anatomy_phase", reasoners.AnatomyPhase) - regReasoner(n, "planning_phase", reasoners.PlanningPhase) - regReasoner(n, "meta_semantic", reasoners.MetaSemantic) - regReasoner(n, "meta_mechanical", reasoners.MetaMechanical) - regReasoner(n, "meta_systemic", reasoners.MetaSystemic) - regReasoner(n, "review_dimension", reasoners.ReviewDimension) - regReasoner(n, "compound_finder_phase", reasoners.CompoundFinderPhase) - regReasoner(n, "post_worthiness_gate", reasoners.PostWorthinessGate) - regReasoner(n, "compound_dedup_phase", reasoners.CompoundDedupPhase) - regReasoner(n, "evidence_verifier", reasoners.EvidenceVerifier) - regReasoner(n, "adversary_phase", reasoners.AdversaryPhase) - regReasoner(n, "deepen_findings", reasoners.DeepenFindings) - regReasoner(n, "extract_obligations", reasoners.ExtractObligations) - regReasoner(n, "verify_obligation", reasoners.VerifyObligation) - regReasoner(n, "coverage_gate", reasoners.CoverageGate) + // from the agent (Harness=App, AI=App). Names come from the reasoners.Name* + // constants — the same ones the orchestrator's CallLocal-routed seams invoke + // (orch.callLocalSeams), so the DAG phase names cannot drift from the + // registered surface. + regReasoner(n, reasoners.NameIntakePhase, reasoners.IntakePhase) + regReasoner(n, reasoners.NameAnatomyPhase, reasoners.AnatomyPhase) + regReasoner(n, reasoners.NamePlanningPhase, reasoners.PlanningPhase) + regReasoner(n, reasoners.NameMetaSemantic, reasoners.MetaSemantic) + regReasoner(n, reasoners.NameMetaMechanical, reasoners.MetaMechanical) + regReasoner(n, reasoners.NameMetaSystemic, reasoners.MetaSystemic) + regReasoner(n, reasoners.NameReviewDimension, reasoners.ReviewDimension) + regReasoner(n, reasoners.NameCompoundFinderPhase, reasoners.CompoundFinderPhase) + regReasoner(n, reasoners.NamePostWorthinessGate, reasoners.PostWorthinessGate) + regReasoner(n, reasoners.NameCompoundDedupPhase, reasoners.CompoundDedupPhase) + regReasoner(n, reasoners.NameEvidenceVerifier, reasoners.EvidenceVerifier) + regReasoner(n, reasoners.NameAdversaryPhase, reasoners.AdversaryPhase) + regReasoner(n, reasoners.NameDeepenFindings, reasoners.DeepenFindings) + regReasoner(n, reasoners.NameExtractObligations, reasoners.ExtractObligations) + regReasoner(n, reasoners.NameVerifyObligation, reasoners.VerifyObligation) + regReasoner(n, reasoners.NameCoverageGate, reasoners.CoverageGate) } // record appends name (and its tags) to the node's registration bookkeeping — @@ -131,6 +134,7 @@ func (n *Node) reviewHandler(ctx context.Context, input map[string]any) (any, er GH: n.gh, NodeID: n.NodeID, AgentFieldServer: n.AgentFieldServer, + Local: n.localCaller, } result, err := n.runReview(ctx, deps, in, cfg) diff --git a/go/internal/orch/calllocal_test.go b/go/internal/orch/calllocal_test.go new file mode 100644 index 0000000..24fb1a7 --- /dev/null +++ b/go/internal/orch/calllocal_test.go @@ -0,0 +1,178 @@ +package orch + +// DAG-routing parity: when Deps.Local is wired (production), every phase seam +// must invoke the SDK's tracked CallLocal under its registered reasoner name — +// that is what makes each phase a child execution on the control plane and +// renders the same pipeline DAG the Python port produces. Contract items: +// +// 1. each seam routes through CallLocal with the exact registered name; +// 2. the input map CallLocal receives afx.Binds back into the identical typed +// input the orchestrator built (lossless round trip — including deliberate +// zero values on fields whose Bind-side defaults are non-zero); +// 3. the handler's map[string]any result passes through unchanged; +// 4. a CallLocal error propagates unchanged; +// 5. Deps.Local == nil keeps the plain direct-call seams (stub-based tests +// and the pre-DAG behavior). + +import ( + "context" + "errors" + "reflect" + "testing" + + "github.com/Agent-Field/pr-af/go/internal/afx" + "github.com/Agent-Field/pr-af/go/internal/config" + "github.com/Agent-Field/pr-af/go/internal/reasoners" + "github.com/Agent-Field/pr-af/go/internal/schemas" +) + +// fakeLocal records the last CallLocal invocation and returns a scripted +// result/error. +type fakeLocal struct { + names []string + name string + input map[string]any + result any + err error +} + +func (f *fakeLocal) CallLocal(_ context.Context, name string, input map[string]any) (any, error) { + f.names = append(f.names, name) + f.name = name + f.input = input + return f.result, f.err +} + +// checkSeam drives one CallLocal-routed seam and asserts contract items 1-3. +func checkSeam[T any]( + t *testing.T, + fake *fakeLocal, + seam func(context.Context, reasoners.Deps, T) (map[string]any, error), + wantName string, + in T, +) { + t.Helper() + want := map[string]any{"ok": true, "phase": wantName} + fake.result, fake.err = want, nil + + got, err := seam(context.Background(), reasoners.Deps{}, in) + if err != nil { + t.Fatalf("%s: seam error: %v", wantName, err) + } + if fake.name != wantName { + t.Fatalf("seam called CallLocal(%q), want %q", fake.name, wantName) + } + if !reflect.DeepEqual(got, want) { + t.Errorf("%s: result = %v, want the handler map unchanged", wantName, got) + } + + back, err := afx.Bind[T](fake.input) + if err != nil { + t.Fatalf("%s: input map does not Bind back: %v", wantName, err) + } + if !reflect.DeepEqual(back, in) { + t.Errorf("%s: round trip mutated the input:\n sent: %#v\n back: %#v", wantName, in, back) + } +} + +func TestCallLocalSeamsRouteEveryPhase(t *testing.T) { + fake := &fakeLocal{} + o := New(Deps{App: &fakeApp{}, Local: fake}, schemas.ReviewInput{}, config.DefaultReviewConfig()) + + intake := schemas.IntakeResult{PrSummary: "s", Complexity: "standard"} + anatomy := schemas.AnatomyResult{} + patches := reasoners.OrderedPatches{{Key: "b.go", Val: "@@ -1 +1 @@"}, {Key: "a.go", Val: "@@ -2 +2 @@"}} + + checkSeam(t, fake, o.rfns.intake, reasoners.NameIntakePhase, + reasoners.IntakeInput{PRData: schemas.GitHubPRData{Title: "t"}, Depth: "deep"}) + checkSeam(t, fake, o.rfns.anatomy, reasoners.NameAnatomyPhase, + reasoners.AnatomyInput{Intake: intake, RepoPath: "/repo"}) + checkSeam(t, fake, o.rfns.metaSemantic, reasoners.NameMetaSemantic, + reasoners.MetaInput{Intake: intake, Anatomy: anatomy, Depth: "standard", DiffPatches: patches}) + checkSeam(t, fake, o.rfns.metaMechanical, reasoners.NameMetaMechanical, + reasoners.MetaInput{Depth: "quick"}) + checkSeam(t, fake, o.rfns.metaSystemic, reasoners.NameMetaSystemic, + reasoners.MetaInput{ReviewerFeedback: "fb"}) + // CurrentDepth=1 with a deliberate MaxDepth=0: Bind seeds MaxDepth=2 only + // for an ABSENT key, and ToMap always emits the key — a zero the + // orchestrator sets must survive the round trip (contract item 2). + checkSeam(t, fake, o.rfns.reviewDim, reasoners.NameReviewDimension, + reasoners.ReviewDimensionInput{ + ReviewPrompt: "look", TargetFiles: []string{"a.go"}, CurrentDepth: 1, MaxDepth: 0, + DiffPatches: map[string]string{"a.go": "@@"}, + }) + checkSeam(t, fake, o.rfns.postWorthiness, reasoners.NamePostWorthinessGate, + reasoners.PostWorthinessInput{Findings: []schemas.ReviewFinding{}}) + checkSeam(t, fake, o.rfns.evidenceVerify, reasoners.NameEvidenceVerifier, + reasoners.EvidenceVerifierInput{PrContext: "ctx"}) + checkSeam(t, fake, o.rfns.adversary, reasoners.NameAdversaryPhase, + reasoners.AdversaryInput{AIGeneratedConfidence: 0.7}) + checkSeam(t, fake, o.rfns.compoundFinder, reasoners.NameCompoundFinderPhase, + reasoners.CompoundFinderInput{RepoPath: "/repo"}) + checkSeam(t, fake, o.rfns.compoundDedup, reasoners.NameCompoundDedupPhase, + reasoners.CompoundDedupInput{IndividualFindingsSummary: "sum"}) + checkSeam(t, fake, o.rfns.coverageGate, reasoners.NameCoverageGate, + reasoners.CoverageGateInput{}) + checkSeam(t, fake, o.rfns.extractOblig, reasoners.NameExtractObligations, + reasoners.ExtractObligationsInput{DiffPatches: patches}) + checkSeam(t, fake, o.rfns.verifyOblig, reasoners.NameVerifyObligation, + reasoners.VerifyObligationInput{RepoPath: "/repo"}) + + // Every invoked name must be one the node registers on the control plane — + // the DAG's phase nodes carry these exact reasoner ids. + wantNames := []string{ + reasoners.NameIntakePhase, reasoners.NameAnatomyPhase, + reasoners.NameMetaSemantic, reasoners.NameMetaMechanical, reasoners.NameMetaSystemic, + reasoners.NameReviewDimension, reasoners.NamePostWorthinessGate, + reasoners.NameEvidenceVerifier, reasoners.NameAdversaryPhase, + reasoners.NameCompoundFinderPhase, reasoners.NameCompoundDedupPhase, + reasoners.NameCoverageGate, reasoners.NameExtractObligations, reasoners.NameVerifyObligation, + } + if !reflect.DeepEqual(fake.names, wantNames) { + t.Errorf("CallLocal names = %v, want %v", fake.names, wantNames) + } +} + +func TestCallLocalSeamPropagatesError(t *testing.T) { + fake := &fakeLocal{err: errors.New("harness exploded")} + o := New(Deps{App: &fakeApp{}, Local: fake}, schemas.ReviewInput{}, config.DefaultReviewConfig()) + + _, err := o.rfns.intake(context.Background(), reasoners.Deps{}, reasoners.IntakeInput{}) + if err == nil || err.Error() != "harness exploded" { + t.Fatalf("err = %v, want the CallLocal error unchanged", err) + } +} + +func TestCallLocalSeamRejectsNonMapResult(t *testing.T) { + fake := &fakeLocal{result: "not a map"} + o := New(Deps{App: &fakeApp{}, Local: fake}, schemas.ReviewInput{}, config.DefaultReviewConfig()) + + _, err := o.rfns.intake(context.Background(), reasoners.Deps{}, reasoners.IntakeInput{}) + if err == nil { + t.Fatal("expected an error for a non-map CallLocal result") + } +} + +func TestCallLocalSeamNilResultYieldsEmptyMap(t *testing.T) { + fake := &fakeLocal{} + o := New(Deps{App: &fakeApp{}, Local: fake}, schemas.ReviewInput{}, config.DefaultReviewConfig()) + + got, err := o.rfns.intake(context.Background(), reasoners.Deps{}, reasoners.IntakeInput{}) + if err != nil { + t.Fatalf("seam error: %v", err) + } + if got == nil || len(got) != 0 { + t.Fatalf("got %v, want an empty non-nil map", got) + } +} + +func TestNilLocalKeepsDirectSeams(t *testing.T) { + o := New(Deps{App: &fakeApp{}}, schemas.ReviewInput{}, config.DefaultReviewConfig()) + + if reflect.ValueOf(o.rfns.intake).Pointer() != reflect.ValueOf(reasoners.IntakePhase).Pointer() { + t.Error("intake seam is not the direct reasoners.IntakePhase when Local is nil") + } + if reflect.ValueOf(o.rfns.verifyOblig).Pointer() != reflect.ValueOf(reasoners.VerifyObligation).Pointer() { + t.Error("verifyOblig seam is not the direct reasoners.VerifyObligation when Local is nil") + } +} diff --git a/go/internal/orch/orchestrator.go b/go/internal/orch/orchestrator.go index 2b0887a..de68cce 100644 --- a/go/internal/orch/orchestrator.go +++ b/go/internal/orch/orchestrator.go @@ -26,6 +26,7 @@ import ( "github.com/Agent-Field/agentfield/sdk/go/ai" "github.com/Agent-Field/agentfield/sdk/go/harness" + "github.com/Agent-Field/pr-af/go/internal/afx" "github.com/Agent-Field/pr-af/go/internal/config" "github.com/Agent-Field/pr-af/go/internal/github" "github.com/Agent-Field/pr-af/go/internal/hitl" @@ -63,6 +64,22 @@ type App interface { // Compile-time proof the live agent satisfies the surface. var _ App = (*agent.Agent)(nil) +// LocalCaller is the SDK surface for same-process reasoner invocation with +// workflow tracking (Agent.CallLocal): each call builds a child execution +// context from ctx and emits running/succeeded/failed events to the control +// plane, so every pipeline phase renders as its own node in the run's DAG — +// the same orchestration graph the Python port produces via its tracked +// router-reasoner calls. Python's @router.reasoner() wrapper routes direct +// in-process calls through workflow instrumentation; CallLocal is the Go +// SDK's equivalent, and routing the phase seams through it is what keeps the +// Go node from collapsing into a single opaque `review` execution. +type LocalCaller interface { + CallLocal(ctx context.Context, reasonerName string, input map[string]any) (any, error) +} + +// Compile-time proof the live agent satisfies the local-call surface. +var _ LocalCaller = (*agent.Agent)(nil) + // Deps carries the injected capabilities. Divergence from design §C.6: the hax // client is NOT a Deps field — Python builds it inside run() via // build_hax_client_from_env (gated on pr_url && !dry_run), so the orchestrator @@ -74,6 +91,13 @@ type Deps struct { GH github.Client NodeID string AgentFieldServer string + + // Local, when non-nil, routes every phase invocation through + // LocalCaller.CallLocal so the control plane records one child execution + // per phase (the pipeline DAG). nil falls back to plain in-process + // function calls — the pre-DAG behavior unit tests and stub harnesses + // rely on. Production (node/register.go) always wires the live agent. + Local LocalCaller } // phaseOrder ports ReviewOrchestrator.PHASE_ORDER — the cost-breakdown / @@ -184,6 +208,59 @@ func defaultReasonerSeams() reasonerSeams { } } +// callLocalSeams routes every phase through local.CallLocal under its +// registered reasoner name (reasoners.Name*), so each invocation is tracked +// as a child execution on the control plane. The registered handler +// (node/register.go) afx.Binds the map back into the same typed input and +// calls the same reasoners.* function with the same Deps the direct seams +// use — behavior is identical, the DAG is the only addition. +func callLocalSeams(local LocalCaller) reasonerSeams { + return reasonerSeams{ + intake: viaLocal[reasoners.IntakeInput](local, reasoners.NameIntakePhase), + anatomy: viaLocal[reasoners.AnatomyInput](local, reasoners.NameAnatomyPhase), + metaSemantic: viaLocal[reasoners.MetaInput](local, reasoners.NameMetaSemantic), + metaMechanical: viaLocal[reasoners.MetaInput](local, reasoners.NameMetaMechanical), + metaSystemic: viaLocal[reasoners.MetaInput](local, reasoners.NameMetaSystemic), + reviewDim: viaLocal[reasoners.ReviewDimensionInput](local, reasoners.NameReviewDimension), + postWorthiness: viaLocal[reasoners.PostWorthinessInput](local, reasoners.NamePostWorthinessGate), + evidenceVerify: viaLocal[reasoners.EvidenceVerifierInput](local, reasoners.NameEvidenceVerifier), + adversary: viaLocal[reasoners.AdversaryInput](local, reasoners.NameAdversaryPhase), + compoundFinder: viaLocal[reasoners.CompoundFinderInput](local, reasoners.NameCompoundFinderPhase), + compoundDedup: viaLocal[reasoners.CompoundDedupInput](local, reasoners.NameCompoundDedupPhase), + coverageGate: viaLocal[reasoners.CoverageGateInput](local, reasoners.NameCoverageGate), + extractOblig: viaLocal[reasoners.ExtractObligationsInput](local, reasoners.NameExtractObligations), + verifyOblig: viaLocal[reasoners.VerifyObligationInput](local, reasoners.NameVerifyObligation), + } +} + +// viaLocal adapts one typed seam to a CallLocal invocation: typed input → +// afx.ToMap → CallLocal(name) → registered handler (afx.Bind → reasoners.*). +// The reasoners.Deps parameter is ignored — the handler closes over the +// node-level Deps built at registration, which point at the same live agent. +// Handlers return map[string]any; a nil result (impossible today: every +// reasoner returns a non-nil map or an error) surfaces as an empty map so +// callers keep their raw-map contract. +func viaLocal[T any](local LocalCaller, name string) func(context.Context, reasoners.Deps, T) (map[string]any, error) { + return func(ctx context.Context, _ reasoners.Deps, in T) (map[string]any, error) { + input, err := afx.ToMap(in) + if err != nil { + return nil, err + } + raw, err := local.CallLocal(ctx, name, input) + if err != nil { + return nil, err + } + out, ok := raw.(map[string]any) + if !ok { + if raw == nil { + return map[string]any{}, nil + } + return nil, fmt.Errorf("orch: reasoner %s returned %T, want map[string]any", name, raw) + } + return out, nil + } +} + // New constructs an Orchestrator with production seams. startedAt is set now so // the wall-clock budget gate measures from construction (parity with Python's // time.monotonic() in __init__). @@ -201,6 +278,9 @@ func New(d Deps, in schemas.ReviewInput, cfg config.ReviewConfig) *Orchestrator for _, p := range phaseOrder { o.costBreakdown[p] = 0.0 } + if d.Local != nil { + o.rfns = callLocalSeams(d.Local) + } o.clock = func() time.Duration { return time.Since(o.startedAt) } o.runIntakeFn = o.runIntake diff --git a/go/internal/reasoners/names.go b/go/internal/reasoners/names.go new file mode 100644 index 0000000..d75cd77 --- /dev/null +++ b/go/internal/reasoners/names.go @@ -0,0 +1,25 @@ +package reasoners + +// Control-plane registration names of the 16 router reasoners (design §B.1). +// node/register.go registers each handler under these names, and the +// orchestrator's CallLocal-routed seams (orch.callLocalSeams) invoke them by +// the same constants — a single source of truth so the DAG's phase names can +// never drift from the registered surface. +const ( + NameIntakePhase = "intake_phase" + NameAnatomyPhase = "anatomy_phase" + NamePlanningPhase = "planning_phase" + NameMetaSemantic = "meta_semantic" + NameMetaMechanical = "meta_mechanical" + NameMetaSystemic = "meta_systemic" + NameReviewDimension = "review_dimension" + NameCompoundFinderPhase = "compound_finder_phase" + NamePostWorthinessGate = "post_worthiness_gate" + NameCompoundDedupPhase = "compound_dedup_phase" + NameEvidenceVerifier = "evidence_verifier" + NameAdversaryPhase = "adversary_phase" + NameDeepenFindings = "deepen_findings" + NameExtractObligations = "extract_obligations" + NameVerifyObligation = "verify_obligation" + NameCoverageGate = "coverage_gate" +) From 52a6fe1fa6579cb866d7eb4704dfd811c171aeab Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Tue, 14 Jul 2026 16:10:28 -0400 Subject: [PATCH 4/4] docs(go): document the pipeline DAG on the control plane Co-Authored-By: Claude Fable 5 --- go/README.md | 41 ++++++++++++++++++++++++++++++++++++++--- 1 file changed, 38 insertions(+), 3 deletions(-) diff --git a/go/README.md b/go/README.md index 9ddbaf4..cfc347c 100644 --- a/go/README.md +++ b/go/README.md @@ -1,9 +1,12 @@ # PR-AF — Go node A Go port of the PR-AF agentic code-review node. It registers the same reasoner -surface under the same names as the Python node and exposes a byte-compatible -HTTP API, so the control-plane DAG UI renders identically. The Python package -under `src/pr_af/` is untouched; this port lives entirely under `go/`. +surface under the same names as the Python node, exposes a byte-compatible +HTTP API, and reports every pipeline phase as its own tracked execution, so the +control-plane DAG UI renders the same multi-node orchestration graph as the +Python node (see [Pipeline DAG on the control plane](#pipeline-dag-on-the-control-plane)). +The Python package under `src/pr_af/` is untouched; this port lives entirely +under `go/`. One binary: @@ -28,6 +31,38 @@ curl -X POST http://localhost:8080/api/v1/execute/async/pr-af-go.review \ `NODE_ID` / `PORT` still override the defaults if you want a different id/port. +## Pipeline DAG on the control plane + +The Python node's review is not one execution — every phase +(`intake_phase`, `anatomy_phase`, the three `meta_*` selectors, each parallel +`review_dimension`, `adversary_phase`, `evidence_verifier`, the obligation and +compound reasoners, `coverage_gate`) runs as a **tracked child execution** of +the parent `review`, and the control-plane UI renders the run as that DAG. +Python gets this from its `@router.reasoner()` wrapper, which routes direct +in-process calls through workflow instrumentation. + +The Go port reproduces the same graph through the SDK's `Agent.CallLocal`: +the orchestrator's phase seams (`orch.callLocalSeams`) invoke each phase under +its registered reasoner name instead of calling the function directly. Each +call builds a child execution context from the incoming request's context and +emits `running` / `succeeded` / `failed` workflow events to the control plane, +which mirrors them into the executions table — one DAG node per phase, parented +under the `review` execution, same shape as Python. + +Mechanics worth knowing: + +- **Same code path either way.** A CallLocal-routed phase goes through the + registered handler (`afx.Bind` into the typed input → the same `reasoners.*` + function with the same deps). `afx.ToMap` keeps nested values typed so + custom marshalers (notably `OrderedPatches`' insertion-ordered + `diff_patches`) survive the round trip byte-for-byte. +- **Best-effort reporting.** Workflow events are fire-and-forget: an + unreachable control plane logs a warning and the review proceeds — the DAG + is observability, never a failure mode. +- **Stubs opt out.** `orch.Deps.Local` is nil in unit tests and stub + harnesses, which keeps the plain direct-call seams; production wiring + (`node.BuildAgent`) always points it at the live agent. + ## Depending on the AgentField Go SDK Unlike the SWE-AF Go port, this module depends on the AgentField Go SDK