fix(audit): per-invocation attribution for egress-proxy + MCP consent events (#341, #366)

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**,

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 |

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),
 		})

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,
 				},
 			})
 		}

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) {

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.

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)
+	}
+}

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
+}