From 977533e4d54da8f3a68d97a4eb336b47639d6cad Mon Sep 17 00:00:00 2001 From: MK Date: Fri, 24 Jul 2026 14:12:29 -0400 Subject: [PATCH 1/3] fix(audit): per-invocation seq for egress-proxy events (#341) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Egress-proxy audit events (source=proxy) carried task_id/correlation_id (#338) but no seq, so they were invisible to gap-detection (which walks seq within a (correlation_id, task_id) group). Root cause: the proxy is a separate 127.0.0.1 forwarder with no request ctx, so it emits via plain Emit and skips the ctx-scoped counter. Add coreruntime.SequenceRegistry — a per-invocation counter registry keyed by (correlation_id, task_id). The runner registers each invocation's counter at request entry (all three A2A entry points) and evicts at the invocation boundary; the proxy OnAttempt recovers (corr, task) from the replayed Proxy-Authorization creds and advances the SAME counter via NextSequenceFor, so proxy events join the gap-free seq chain. A miss returns 0 (seq-less, never a wrong/duplicate seq) — startup + unattributed events are unaffected. This registry is also the seam #366 uses to attribute MCP consent-resume egress. Tests: registry Register/Get/Evict/NextSequenceFor + shared-counter + race + nil-safe; runner registerInvocationSeq shares the in-context counter and evicts. Build/vet/lint clean. --- forge-cli/runtime/runner.go | 29 ++++++- forge-cli/runtime/seq_registry_wiring_test.go | 43 +++++++++++ forge-core/runtime/audit_schema.go | 74 ++++++++++++++++++ forge-core/runtime/sequence_registry_test.go | 76 +++++++++++++++++++ 4 files changed, 221 insertions(+), 1 deletion(-) create mode 100644 forge-cli/runtime/seq_registry_wiring_test.go create mode 100644 forge-core/runtime/sequence_registry_test.go diff --git a/forge-cli/runtime/runner.go b/forge-cli/runtime/runner.go index 80149290..0443ab85 100644 --- a/forge-cli/runtime/runner.go +++ b/forge-cli/runtime/runner.go @@ -171,6 +171,7 @@ type Runner struct { deferralNotifier DeferralNotifier // optional: delivers DEFER approval requests to channels (#310) authToken string // resolved auth token (empty if --no-auth) cancelRegistry *coreruntime.CancellationRegistry // per-Runner in-flight cancellation registry (issue #88 / FWS-4) + seqRegistry *coreruntime.SequenceRegistry // per-invocation seq counters keyed by (correlation_id, task_id) so out-of-band emitters (egress proxy #341, MCP consent-resume #366) stamp a correct seq auditSigningKey *coreruntime.LoadedKey // loaded once at startup; nil when signing is off (#213). Served on JWKS endpoint. compression *compress.Runtime // ctxzip compression runtime; nil when compression is disabled intentEngine *intent.Engine // R3 (#208) intent-alignment engine; nil when disabled @@ -205,6 +206,7 @@ func NewRunner(cfg RunnerConfig) (*Runner, error) { cfg: cfg, logger: logger, cancelRegistry: coreruntime.NewCancellationRegistry(), + seqRegistry: coreruntime.NewSequenceRegistry(), }, nil } @@ -770,7 +772,13 @@ func (r *Runner) Run(ctx context.Context) error { Event: event, TaskID: a.TaskID, CorrelationID: a.CorrelationID, - Fields: map[string]any{"domain": a.Domain, "mode": string(egressCfg.Mode), "source": "proxy"}, + // #341 — recover the invocation's live seq counter by the + // (correlation_id, task_id) the subprocess replayed (#338), + // so proxy events join the same gap-detectable seq chain as + // in-process egress. 0 (omitempty) when the invocation isn't + // registered (startup, or a subprocess with no creds). + Sequence: r.seqRegistry.NextSequenceFor(a.CorrelationID, a.TaskID), + Fields: map[string]any{"domain": a.Domain, "mode": string(egressCfg.Mode), "source": "proxy"}, }) } var pErr error @@ -1644,6 +1652,8 @@ func (r *Runner) registerHandlers(srv *server.Server, executor coreruntime.Agent // EnsureSequenceCounter installs a fresh one if missing // (--no-auth path / direct test invocations). ctx = coreruntime.EnsureSequenceCounter(ctx) + // #341/#366: expose the seq counter to out-of-band emitters (egress proxy, MCP consent-resume). + defer r.registerInvocationSeq(ctx)() sseAcc := coreruntime.NewLLMUsageAccumulator() ctx = coreruntime.WithLLMUsageAccumulator(ctx, sseAcc) defer func() { @@ -1848,6 +1858,19 @@ func (r *Runner) registerHandlers(srv *server.Server, executor coreruntime.Agent }) } +// registerInvocationSeq exposes this invocation's sequence counter by +// (correlation_id, task_id) so events emitted outside the request goroutine — +// the egress proxy (#341) and the MCP consent-resume paths (#366) — can advance +// the SAME counter and stamp a correct, gap-free seq. Call it after the +// correlation id, task id, and sequence counter are on ctx; defer the returned +// evict closure so the registration is dropped at the invocation boundary. +func (r *Runner) registerInvocationSeq(ctx context.Context) func() { + corr := coreruntime.CorrelationIDFromContext(ctx) + task := coreruntime.TaskIDFromContext(ctx) + r.seqRegistry.Register(corr, task, coreruntime.SequenceCounterFromContext(ctx)) + return func() { r.seqRegistry.Evict(corr, task) } +} + // executeTask is the shared task execution pipeline used by both JSON-RPC and REST handlers. func (r *Runner) executeTask( ctx context.Context, @@ -1870,6 +1893,8 @@ func (r *Runner) executeTask( // session_start lands seq=2 (#174); installs a fresh one when // missing (--no-auth path / direct test invocations). ctx = coreruntime.EnsureSequenceCounter(ctx) + // #341/#366: expose the seq counter to out-of-band emitters (egress proxy, MCP consent-resume). + defer r.registerInvocationSeq(ctx)() // Per-invocation usage accumulator so AfterLLMCall hooks can fold // each call's tokens/duration into running totals the response // handler reads back for X-Forge-* headers + the @@ -2181,6 +2206,8 @@ func (r *Runner) registerRESTHandlers(srv *server.Server, executor coreruntime.A // installSequenceCounterMiddleware put on ctx before auth ran // (#174); install fresh on the --no-auth path. ctx = coreruntime.EnsureSequenceCounter(ctx) + // #341/#366: expose the seq counter to out-of-band emitters (egress proxy, MCP consent-resume). + defer r.registerInvocationSeq(ctx)() // Pull workflow correlation headers (issue #86 / FWS-2) before // the accumulator setup so invocation_complete inherits workflow // tagging via EmitFromContext. diff --git a/forge-cli/runtime/seq_registry_wiring_test.go b/forge-cli/runtime/seq_registry_wiring_test.go new file mode 100644 index 00000000..e9e8fab2 --- /dev/null +++ b/forge-cli/runtime/seq_registry_wiring_test.go @@ -0,0 +1,43 @@ +package runtime + +import ( + "context" + "testing" + + coreruntime "github.com/initializ/forge/forge-core/runtime" + "github.com/initializ/forge/forge-core/types" +) + +// TestRegisterInvocationSeq_SharesCounter is the #341/#366 wiring: after +// registerInvocationSeq, an out-of-band emitter (the egress proxy / consent +// resume) that looks up the counter by (correlation_id, task_id) advances the +// SAME counter the in-context request goroutine uses — one gap-free sequence — +// and eviction drops the registration. +func TestRegisterInvocationSeq_SharesCounter(t *testing.T) { + r, err := NewRunner(RunnerConfig{Config: &types.ForgeConfig{AgentID: "agent", Version: "0.1.0", Framework: "forge"}}) + if err != nil { + t.Fatalf("NewRunner: %v", err) + } + + ctx := context.Background() + ctx = coreruntime.WithCorrelationID(ctx, "corr-1") + ctx = coreruntime.WithTaskID(ctx, "task-1") + ctx = coreruntime.EnsureSequenceCounter(ctx) + + if n := coreruntime.NextSequence(ctx); n != 1 { // in-context (e.g. session_start) + t.Fatalf("in-context NextSequence = %d, want 1", n) + } + + evict := r.registerInvocationSeq(ctx) + if n := r.seqRegistry.NextSequenceFor("corr-1", "task-1"); n != 2 { // out-of-band (proxy) + t.Fatalf("out-of-band NextSequenceFor = %d, want 2 (shared counter)", n) + } + if n := coreruntime.NextSequence(ctx); n != 3 { // back in-context, no gap + t.Fatalf("in-context NextSequence = %d, want 3", n) + } + + evict() + if n := r.seqRegistry.NextSequenceFor("corr-1", "task-1"); n != 0 { + t.Errorf("after evict NextSequenceFor = %d, want 0 (unregistered)", n) + } +} diff --git a/forge-core/runtime/audit_schema.go b/forge-core/runtime/audit_schema.go index fd0a76ac..8a461a64 100644 --- a/forge-core/runtime/audit_schema.go +++ b/forge-core/runtime/audit_schema.go @@ -2,6 +2,7 @@ package runtime import ( "context" + "sync" "sync/atomic" ) @@ -85,3 +86,76 @@ func EnsureSequenceCounter(ctx context.Context) context.Context { } return WithSequenceCounter(ctx, new(SequenceCounter)) } + +// SequenceRegistry maps a live per-invocation SequenceCounter to its +// (correlation_id, task_id) key so events emitted OUTSIDE the request +// goroutine's context can advance the SAME counter and stamp a correct, +// gap-free seq. Two emitters need this: +// +// - The egress proxy (#341): a separate 127.0.0.1 forward proxy with no +// request ctx. It recovers (correlation_id, task_id) from the subprocess +// Proxy-Authorization creds (#338) and looks the counter up here. +// - The MCP consent-resume paths (#366): the loopback OAuth callback and the +// platform POST /mcp/consent run on a detached browser/platform request; +// they recover the parked call's (correlation_id, task_id) and seed a ctx +// from the registered counter so the completion egress is seq'd + attributed. +// +// The counter is registered at request entry (alongside EnsureSequenceCounter) +// and evicted at invocation_complete. A miss returns 0 (the event stays +// seq-less rather than carrying a wrong or duplicated number). +type SequenceRegistry struct { + mu sync.Mutex + m map[string]*SequenceCounter +} + +// NewSequenceRegistry returns an empty registry. +func NewSequenceRegistry() *SequenceRegistry { + return &SequenceRegistry{m: make(map[string]*SequenceCounter)} +} + +func seqRegistryKey(correlationID, taskID string) string { + return correlationID + "\x00" + taskID +} + +// Register records the counter under (correlationID, taskID). No-op on a nil +// registry/counter or an all-empty key. +func (r *SequenceRegistry) Register(correlationID, taskID string, c *SequenceCounter) { + if r == nil || c == nil || (correlationID == "" && taskID == "") { + return + } + r.mu.Lock() + r.m[seqRegistryKey(correlationID, taskID)] = c + r.mu.Unlock() +} + +// Get returns the counter for (correlationID, taskID), or nil if none is +// registered. +func (r *SequenceRegistry) Get(correlationID, taskID string) *SequenceCounter { + if r == nil { + return nil + } + r.mu.Lock() + c := r.m[seqRegistryKey(correlationID, taskID)] + r.mu.Unlock() + return c +} + +// Evict drops the registration for (correlationID, taskID). +func (r *SequenceRegistry) Evict(correlationID, taskID string) { + if r == nil { + return + } + r.mu.Lock() + delete(r.m, seqRegistryKey(correlationID, taskID)) + r.mu.Unlock() +} + +// NextSequenceFor advances the registered counter for (correlationID, taskID) +// and returns the new seq, or 0 when no counter is registered (so the caller +// JSON-omits the field rather than emitting a wrong/duplicate seq). +func (r *SequenceRegistry) NextSequenceFor(correlationID, taskID string) int64 { + if c := r.Get(correlationID, taskID); c != nil { + return c.Add(1) + } + return 0 +} diff --git a/forge-core/runtime/sequence_registry_test.go b/forge-core/runtime/sequence_registry_test.go new file mode 100644 index 00000000..2103544f --- /dev/null +++ b/forge-core/runtime/sequence_registry_test.go @@ -0,0 +1,76 @@ +package runtime + +import ( + "context" + "sync" + "testing" +) + +func TestSequenceRegistry_RegisterGetEvict(t *testing.T) { + reg := NewSequenceRegistry() + c := new(SequenceCounter) + reg.Register("corr-1", "task-1", c) + + if got := reg.Get("corr-1", "task-1"); got != c { + t.Fatalf("Get returned %v, want the registered counter", got) + } + if got := reg.Get("corr-1", "other"); got != nil { + t.Errorf("Get for an unregistered key = %v, want nil", got) + } + + reg.Evict("corr-1", "task-1") + if got := reg.Get("corr-1", "task-1"); got != nil { + t.Errorf("Get after Evict = %v, want nil", got) + } +} + +// TestSequenceRegistry_NextSequenceForSharesCounter is the crux: the registry +// advances the SAME counter the in-context path (NextSequence) uses, so an +// out-of-band emitter and the request goroutine produce one gap-free sequence. +func TestSequenceRegistry_NextSequenceForSharesCounter(t *testing.T) { + reg := NewSequenceRegistry() + c := new(SequenceCounter) + ctx := WithSequenceCounter(context.Background(), c) + reg.Register("corr", "task", c) + + if n := NextSequence(ctx); n != 1 { // in-context (e.g. an in-process tool) + t.Fatalf("NextSequence = %d, want 1", n) + } + if n := reg.NextSequenceFor("corr", "task"); n != 2 { // out-of-band (proxy) + t.Fatalf("NextSequenceFor = %d, want 2 (shared counter)", n) + } + if n := NextSequence(ctx); n != 3 { + t.Fatalf("NextSequence = %d, want 3", n) + } +} + +func TestSequenceRegistry_MissReturnsZero(t *testing.T) { + reg := NewSequenceRegistry() + if n := reg.NextSequenceFor("nope", "nope"); n != 0 { + t.Errorf("NextSequenceFor on a miss = %d, want 0 (seq-less, not mis-seq'd)", n) + } +} + +func TestSequenceRegistry_NilSafe(t *testing.T) { + var reg *SequenceRegistry // nil registry must be a no-op, not a panic + reg.Register("c", "t", new(SequenceCounter)) + if reg.Get("c", "t") != nil || reg.NextSequenceFor("c", "t") != 0 { + t.Error("nil registry should be inert") + } + reg.Evict("c", "t") +} + +func TestSequenceRegistry_Concurrent(t *testing.T) { + reg := NewSequenceRegistry() + c := new(SequenceCounter) + reg.Register("c", "t", c) + var wg sync.WaitGroup + for i := 0; i < 100; i++ { + wg.Add(1) + go func() { defer wg.Done(); reg.NextSequenceFor("c", "t") }() + } + wg.Wait() + if got := c.Load(); got != 100 { + t.Errorf("counter = %d after 100 concurrent advances, want 100", got) + } +} From cb92dc259f271ae221601b3e05f2136e363aba80 Mon Sep 17 00:00:00 2001 From: MK Date: Fri, 24 Jul 2026 14:23:44 -0400 Subject: [PATCH 2/3] fix(mcp): attribute consent audit + egress to the parked invocation (#366) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two attribution defects on the MCP delegated-consent path: A. mcp_auth_resolved was double-emitted, and 2 of 3 sites dropped attribution. The waking gate (mcpAuthGate.Await) already emits an attributed mcp_auth_resolved (grant) / mcp_auth_timeout (refusal). The resume channels emitted a second, UNATTRIBUTED copy via plain Emit on a different request: - POST /mcp/consent: dropped entirely — Peek guarantees a waiter, so the gate covers it; a resolved event for a refusal (decision=timeout) was misleading. - loopback callback: now emits ONLY when Resolve finds no waiter (a late grant after the original call timed out — genuinely unattributed, no invocation). B. The loopback callback's code->token exchange ran on the browser request ctx, so its egress surfaced unattributed. The parked invocation is still in flight (its seq counter is still registered #341), so: add CorrelationID to authgate.Spec (travels with the first waiter), recover (correlation_id, task_id) via engine.Peek at the callback, and seed the completion ctx (WithCorrelationID/WithTaskID + the registered seq counter). The token exchange (ExchangeCodeCtx) now attributes + seq's to the parked invocation. Not covered here: the pooled MCP connection establishment (subject_pool.go context.Background()) — a connection is shared across invocations, so that egress is infrastructure (like startup), not cleanly per-invocation; documented as a follow-up boundary. Tests: callback seeds the parked (corr,task)+shared counter; authgate + consent suites green. --- forge-cli/runtime/mcp_authgate.go | 17 +++---- forge-cli/runtime/mcp_consent_callback.go | 33 ++++++++++--- .../runtime/mcp_consent_callback_test.go | 48 ++++++++++++++++++- forge-core/security/authgate/authgate.go | 5 ++ 4 files changed, 85 insertions(+), 18 deletions(-) diff --git a/forge-cli/runtime/mcp_authgate.go b/forge-cli/runtime/mcp_authgate.go index 7c419f80..605cb1c8 100644 --- a/forge-cli/runtime/mcp_authgate.go +++ b/forge-cli/runtime/mcp_authgate.go @@ -60,7 +60,7 @@ func (g *mcpAuthGate) Await(ctx context.Context, server string) error { taskID := coreruntime.TaskIDFromContext(ctx) correlationID := coreruntime.CorrelationIDFromContext(ctx) - handle, first, err := g.engine.Await(subject, server, authgate.Spec{TaskID: taskID}) + handle, first, err := g.engine.Await(subject, server, authgate.Spec{TaskID: taskID, CorrelationID: correlationID}) if err != nil { return fmt.Errorf("%w: %v", mcp.ErrNoToken, err) } @@ -221,15 +221,12 @@ func makeMCPConsentHandler(engine *authgate.Engine, auditLogger *coreruntime.Aud writeJSON(w, http.StatusConflict, map[string]string{"error": err.Error()}) return } - if auditLogger != nil { - auditLogger.Emit(coreruntime.AuditEvent{ - Event: coreruntime.EventMCPAuthResolved, - Fields: map[string]any{ - "server": serverName, "subject": subject, - "decision": string(decision), "via": "consent_endpoint", - }, - }) - } + // The waking gate (mcpAuthGate.Await) emits the ATTRIBUTED terminal + // event for the resumed call — mcp_auth_resolved on grant, + // mcp_auth_timeout on refusal. Don't emit a duplicate here (#366): Peek + // above guarantees a waiter existed, and a resolved event for a refusal + // (decision=timeout) would be misleading. The HTTP 200 is the caller's + // confirmation. writeJSON(w, http.StatusOK, map[string]any{ "subject": subject, "server": serverName, "decision": string(decision), }) diff --git a/forge-cli/runtime/mcp_consent_callback.go b/forge-cli/runtime/mcp_consent_callback.go index 5358ad26..ea1a3e1e 100644 --- a/forge-cli/runtime/mcp_consent_callback.go +++ b/forge-cli/runtime/mcp_consent_callback.go @@ -185,7 +185,7 @@ func (r *Runner) registerMCPCallbackEndpoint(srv *server.Server, auditLogger *co } srv.RegisterHTTPHandler("GET /mcp/oauth/start", makeMCPStartHandler(r.stateBinder)) srv.RegisterHTTPHandler("GET /mcp/oauth/callback", - makeMCPCallbackHandler(r.stateBinder, r.authGateEngine, r.callbackCompleter, sessionFromRequest, auditLogger)) + makeMCPCallbackHandler(r.stateBinder, r.authGateEngine, r.callbackCompleter, sessionFromRequest, auditLogger, r.seqRegistry)) } // forgeSessionCookie is the cookie name the callback's cross-session guard @@ -241,6 +241,7 @@ func makeMCPCallbackHandler( complete CallbackCompleter, sessionOf func(*http.Request) string, auditLogger *coreruntime.AuditLogger, + seqRegistry *coreruntime.SequenceRegistry, ) http.HandlerFunc { return func(w http.ResponseWriter, req *http.Request) { q := req.URL.Query() @@ -279,24 +280,42 @@ func makeMCPCallbackHandler( http.Error(w, "state/session mismatch", http.StatusBadRequest) return } + // Attribute the completion egress (code->token exchange) to the parked + // invocation still waiting on this gate (#366). The gate carries the + // invocation's (correlation_id, task_id); its seq counter is still + // registered because the parked call hasn't returned. Seed the ctx so + // the token-exchange egress joins the invocation's audit + seq chain + // instead of surfacing as an unattributed event on the browser request. + ctx := req.Context() + if h, ok := engine.Peek(b.subject, b.server); ok { + sp := h.Spec() + ctx = coreruntime.WithCorrelationID(coreruntime.WithTaskID(ctx, sp.TaskID), sp.CorrelationID) + if c := seqRegistry.Get(sp.CorrelationID, sp.TaskID); c != nil { + ctx = coreruntime.WithSequenceCounter(ctx, c) + } + } // Exchange the code for a token and store it for {subject, server}. // Only AFTER this succeeds do we resume — resolving the gate with no // grant would just re-park the call (delegation follows // authorization: never resume before the grant exists). The PKCE // verifier bound to the state proves this is the same flow we started. - if err := complete(req.Context(), b.subject, b.server, code, b.verifier); err != nil { + if err := complete(ctx, b.subject, b.server, code, b.verifier); err != nil { http.Error(w, "authorization exchange failed", http.StatusBadGateway) return } // Grant exists now → wake every call parked on {subject, server}. - // A missing gate (call already timed out / was canceled) is benign: - // the token is stored, so a fresh call will just succeed. - _ = engine.Resolve(b.subject, b.server, authgate.DecisionGranted) - if auditLogger != nil { + // When a waiter is present the waking gate (mcpAuthGate.Await) emits the + // ATTRIBUTED mcp_auth_resolved for the resumed call — so we don't emit + // here (#366: avoid the unattributed duplicate). A missing gate (the + // original call already timed out / was canceled) is benign — the token + // is stored for a fresh call — but there's no in-flight invocation to + // attribute to, so record that late grant here (unattributed by nature). + if err := engine.Resolve(b.subject, b.server, authgate.DecisionGranted); err != nil && auditLogger != nil { auditLogger.Emit(coreruntime.AuditEvent{ Event: coreruntime.EventMCPAuthResolved, Fields: map[string]any{ - "server": b.server, "subject": b.subject, "via": "loopback_callback", + "server": b.server, "subject": b.subject, + "via": "loopback_callback", "late": true, }, }) } diff --git a/forge-cli/runtime/mcp_consent_callback_test.go b/forge-cli/runtime/mcp_consent_callback_test.go index be285c24..cb18e3a2 100644 --- a/forge-cli/runtime/mcp_consent_callback_test.go +++ b/forge-cli/runtime/mcp_consent_callback_test.go @@ -8,6 +8,7 @@ import ( "testing" "time" + coreruntime "github.com/initializ/forge/forge-core/runtime" "github.com/initializ/forge/forge-core/security/authgate" ) @@ -100,7 +101,7 @@ func (f *fakeCompleter) complete(_ context.Context, _, _, code, _ string) error func newCallback(t *testing.T, binder *stateBinder, engine *authgate.Engine, comp *fakeCompleter) http.HandlerFunc { t.Helper() - return makeMCPCallbackHandler(binder, engine, comp.complete, sessionFromRequest, nil) + return makeMCPCallbackHandler(binder, engine, comp.complete, sessionFromRequest, nil, coreruntime.NewSequenceRegistry()) } // The happy path: valid state + code → exchange runs → the parked call @@ -140,6 +141,51 @@ func TestMCPCallback_HappyPath_ResumesGate(t *testing.T) { } } +// #366: the callback attributes the completion egress to the still-parked +// invocation — the completer's ctx carries the parked (correlation_id, task_id) +// and shares the invocation's registered seq counter. +func TestMCPCallback_SeedsInvocationContext(t *testing.T) { + binder := newStateBinder(time.Hour) + engine := authgate.New() + + // Park a call carrying a known invocation identity (as Await sets it). + handle, _, err := engine.Await("alice@corp.com", "atl", + authgate.Spec{Timeout: time.Hour, TaskID: "task-1", CorrelationID: "corr-1"}) + if err != nil { + t.Fatal(err) + } + go func() { _, _ = handle.WaitCtx(context.Background()) }() + + // The invocation's seq counter is registered while it's parked. + reg := coreruntime.NewSequenceRegistry() + reg.Register("corr-1", "task-1", new(coreruntime.SequenceCounter)) + + var gotCtx context.Context + comp := func(ctx context.Context, _, _, _, _ string) error { gotCtx = ctx; return nil } + + state, _ := binder.Issue("alice@corp.com", "atl", "sess-1") + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", "/mcp/oauth/callback?state="+state+"&code=c", nil) + req.Header.Set("X-Forge-Session", "sess-1") + makeMCPCallbackHandler(binder, engine, comp, sessionFromRequest, nil, reg)(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("callback → %d, want 200", rec.Code) + } + if gotCtx == nil { + t.Fatal("completer was not called") + } + if got := coreruntime.CorrelationIDFromContext(gotCtx); got != "corr-1" { + t.Errorf("completion ctx correlation = %q, want corr-1 (attributed to the parked invocation)", got) + } + if got := coreruntime.TaskIDFromContext(gotCtx); got != "task-1" { + t.Errorf("completion ctx task = %q, want task-1", got) + } + if n := coreruntime.NextSequence(gotCtx); n != 1 { + t.Errorf("completion ctx seq = %d, want 1 (shares the registered invocation counter)", n) + } +} + // Cross-session: a valid state used from a DIFFERENT session is rejected, // the code is NOT exchanged, and no gate resumes. func TestMCPCallback_CrossSessionRejected(t *testing.T) { diff --git a/forge-core/security/authgate/authgate.go b/forge-core/security/authgate/authgate.go index 84f049d8..f6dea113 100644 --- a/forge-core/security/authgate/authgate.go +++ b/forge-core/security/authgate/authgate.go @@ -80,6 +80,11 @@ type Spec struct { // one consent serves them all. TaskID string Session string + // CorrelationID is the invocation id of the request that first tripped the + // gate. It travels with the first waiter so the resume paths (the loopback + // OAuth callback, POST /mcp/consent) can attribute the completion egress + + // audit back to the still-in-flight parked invocation (#366). + CorrelationID string } // Handle is a pending gate. Executors obtain one from Engine.Await and From 8369034f303d8ea0f53c7e405a05fd326decd47f Mon Sep 17 00:00:00 2001 From: MK Date: Fri, 24 Jul 2026 14:25:56 -0400 Subject: [PATCH 3/3] docs: proxy events now carry seq (#341); MCP consent resolved is single + attributed (#366) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit audit-logging.md: update the plain-Emit exception for source=proxy — proxy events now carry a correct seq via the SequenceRegistry (only trace cross-link remains unavailable). delegated-consent.md: mcp_auth_resolved/timeout emitted once by the resuming gate with the invocation's correlation/task/seq; the late-grant no-waiter case is the only unattributed exception. --- docs/mcp/delegated-consent.md | 7 +++++++ docs/security/audit-logging.md | 2 +- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/docs/mcp/delegated-consent.md b/docs/mcp/delegated-consent.md index 24f9d1fe..05cee01f 100644 --- a/docs/mcp/delegated-consent.md +++ b/docs/mcp/delegated-consent.md @@ -166,6 +166,13 @@ When a `type: user` call has no grant, the runtime surfaces the need three ways ``` 3. Resolution emits **`mcp_auth_resolved`**; expiry emits **`mcp_auth_timeout`**. + Both are emitted **once**, by the resuming gate, and carry the parked call's + `correlation_id` / `task_id` / `seq` so they group under the original + invocation (#366) — the resume channels (loopback callback, `POST /mcp/consent`) + no longer emit a second, unattributed copy. The only exception is a *late* + grant that lands after the call already timed out: with no in-flight + invocation to attribute to, the callback records an unattributed + `mcp_auth_resolved{via:loopback_callback,late:true}`. `subject` is the user's email (falls back to the opaque user ID), `server` is the MCP server name, `deadline` is the hard park window — **default 10 minutes**, diff --git a/docs/security/audit-logging.md b/docs/security/audit-logging.md index 5094b32d..33b07d6d 100644 --- a/docs/security/audit-logging.md +++ b/docs/security/audit-logging.md @@ -581,7 +581,7 @@ documented inline: | Site | Why plain `Emit` | |---|---| -| Egress proxy `OnAttempt` with `source=proxy` | Subprocess HTTP `CONNECT` has no Go ctx tying back to the A2A request, so it can't use `EmitFromContext`. Since #338 it still recovers `task_id` + `correlation_id` out-of-band from the `Proxy-Authorization` creds and sets them on the event manually; seq + trace cross-link remain unavailable (no ctx). | +| Egress proxy `OnAttempt` with `source=proxy` | Subprocess HTTP `CONNECT` has no Go ctx tying back to the A2A request, so it can't use `EmitFromContext`. Since #338 it recovers `task_id` + `correlation_id` out-of-band from the `Proxy-Authorization` creds and sets them on the event manually. Since #341 it **also carries a correct `seq`**: the runner registers each invocation's live sequence counter in a `SequenceRegistry` keyed by `(correlation_id, task_id)`, and the proxy `OnAttempt` advances that same counter via `SequenceRegistry.NextSequenceFor` — so proxy events now join the gap-detectable seq chain of their invocation. A miss (startup, or a subprocess with no creds) leaves `seq` at 0 (omitted), never a wrong/duplicate value. Only the trace cross-link (`trace_id`/`span_id`) remains unavailable, since that still needs the ctx's active span. | | MCP server startup events (`mcp_server_started` / `_failed` / `_degraded`) | Pre-invocation; no scope | | Scheduler tick (`schedule_fire` / `schedule_complete` / `schedule_skip` / `schedule_modify`) | Runs on its own timer outside any A2A request | | Startup banners (`policy_loaded`, `agent_card_published`, `audit_export_status`) | Pre-invocation; no scope |