feat(runtime): non-fatal budget exhaustion via tool_result.error (§9.6)

Spec §9.6 SHOULDs the `tool_result.error` form so the agent can decide
whether to continue with non-cost-bearing operations. The runtime
previously always threw `BudgetExhaustedException` from
`Job.EmitMetricAsync`, which the run-loop converted into a terminal
`job.error{BUDGET_EXHAUSTED}` — the agent never got a chance.

- `Job.EmitMetricAsync` now returns the exhausted-currency string (or
  null) instead of throwing, leaving the policy decision to the caller.
- `JobContext.MetricAsync` surfaces the exhaustion as
  `tool_result.error{BUDGET_EXHAUSTED, retryable:false}` and lets the
  agent continue. Opt back into terminal exhaustion with the new
  `ArcpServerOptions.FatalBudgetExhaustion = true`.
- `BudgetEnforcementTests` is rewritten: one test for the new
  spec-preferred surface, one for the legacy fatal opt-in.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: nficano

src/Arcp.Runtime/ArcpServer.cs

@@ -64,7 +64,14 @@ public ArcpServer(ArcpServerOptions options, ILoggerFactory? loggerFactory = nul
                 options.TimeProvider);
         AgentRegistry = new AgentRegistry();
         LeaseManager = new LeaseManager(options.TimeProvider);
-        JobManager = new JobManager(AgentRegistry, LeaseManager, options.TimeProvider, _loggerFactory, CredentialManager, options.IdempotencyWindowSec);
+        JobManager = new JobManager(
+            AgentRegistry,
+            LeaseManager,
+            options.TimeProvider,
+            _loggerFactory,
+            CredentialManager,
+            options.IdempotencyWindowSec,
+            options.FatalBudgetExhaustion);
         if (CredentialManager is not null)
         {
             _ = Task.Run(() => RevokeOutstandingCredentialsAsync(CancellationToken.None));

src/Arcp.Runtime/ArcpServerOptions.cs

@@ -31,6 +31,12 @@ public sealed class ArcpServerOptions
     /// (spec §7.2). Submissions with the same key after this window create a fresh job.</summary>
     public int IdempotencyWindowSec { get; init; } = 3600;
 
+    /// <summary>Whether a <c>cost.budget</c> exhaustion terminates the job with
+    /// <c>BUDGET_EXHAUSTED</c> (legacy v1.0 behavior) or surfaces a non-fatal
+    /// <c>tool_result.error</c> so the agent may emit a partial result and return normally
+    /// (spec §9.6 SHOULD-preferred form). Default: <see langword="false"/>.</summary>
+    public bool FatalBudgetExhaustion { get; init; }
+
     /// <summary>Gets the back pressure threshold.</summary>
     public int BackPressureThreshold { get; init; } = 1000;
 

src/Arcp.Runtime/Job.cs

@@ -214,18 +214,19 @@ internal IReadOnlyList<Envelope> SnapshotEventHistory()
         }
     }
 
-    /// <summary>Apply the budget rule for <c>cost.*</c> metrics, then emit. If the metric exhausts
-    /// the matching budget counter (spec §9.6), the metric event is still emitted but a
-    /// <see cref="BudgetExhaustedException"/> is then thrown so the run-loop terminates the job
-    /// with <c>BUDGET_EXHAUSTED</c> (spec §12).</summary>
-    public async ValueTask EmitMetricAsync(MetricBody body, CancellationToken cancellationToken)
+    /// <summary>Apply the budget rule for <c>cost.*</c> metrics, then emit. Returns the exhausted
+    /// currency (or <see langword="null"/> if all counters remain positive). Callers decide whether
+    /// to surface the exhaustion as a <c>tool_result.error</c> or a fatal exception per spec §9.6
+    /// (which SHOULDs the former).</summary>
+    public async ValueTask<string?> EmitMetricAsync(MetricBody body, CancellationToken cancellationToken)
     {
         var charged = BudgetLedger.ApplyMetric(body.Name, body.Value, body.Unit);
         await EmitEventAsync(EventKinds.Metric, body, cancellationToken).ConfigureAwait(false);
-        if (charged)
+        if (charged && !string.IsNullOrEmpty(body.Unit) && BudgetLedger.IsExhausted(body.Unit))
         {
-            BudgetLedger.AssertNotExhausted();
+            return body.Unit;
         }
+        return null;
     }
 
     /// <summary>Begin result stream.</summary>

src/Arcp.Runtime/JobContext.cs

@@ -23,13 +23,15 @@ public sealed class JobContext
 {
     private readonly Job _job;
     private readonly CredentialManager? _credentials;
+    private readonly bool _fatalBudgetExhaustion;
     private readonly Arcp.Runtime.Leases.LeaseManager? _leases;
 
     internal JobContext(Job job, ILogger logger, CredentialManager? credentials = null,
-                        Arcp.Runtime.Leases.LeaseManager? leases = null)
+                        bool fatalBudgetExhaustion = false, Arcp.Runtime.Leases.LeaseManager? leases = null)
     {
         _job = job;
         _credentials = credentials;
+        _fatalBudgetExhaustion = fatalBudgetExhaustion;
         _leases = leases;
         Logger = logger;
     }
@@ -137,9 +139,34 @@ public ValueTask ToolResultAsync(string callId, object? result, ToolError? error
             Error = error,
         }, cancellationToken);
 
-    /// <summary>Metric (asynchronous).</summary>
-    public ValueTask MetricAsync(string name, double value, string? unit = null, IReadOnlyDictionary<string, string>? dimensions = null, CancellationToken cancellationToken = default) =>
-        _job.EmitMetricAsync(new MetricBody { Name = name, Value = value, Unit = unit, Dimensions = dimensions }, cancellationToken);
+    /// <summary>Emit a <c>metric</c> event. If <paramref name="name"/> begins with <c>cost.</c> and
+    /// <paramref name="unit"/> matches a budgeted currency, the budget counter is decremented
+    /// (spec §9.6). On exhaustion the runtime surfaces a non-fatal <c>tool_result.error</c> with
+    /// code <c>BUDGET_EXHAUSTED</c> so the agent may continue with non-cost-bearing operations
+    /// (spec §9.6 SHOULD); set <see cref="ArcpServerOptions.FatalBudgetExhaustion"/> to opt into
+    /// legacy fatal termination.</summary>
+    public async ValueTask MetricAsync(string name, double value, string? unit = null, IReadOnlyDictionary<string, string>? dimensions = null, CancellationToken cancellationToken = default)
+    {
+        var exhaustedCurrency = await _job
+            .EmitMetricAsync(new MetricBody { Name = name, Value = value, Unit = unit, Dimensions = dimensions }, cancellationToken)
+            .ConfigureAwait(false);
+        if (exhaustedCurrency is null) return;
+
+        var message = $"{exhaustedCurrency} budget exhausted (remaining≤0)";
+        if (_fatalBudgetExhaustion)
+            throw new BudgetExhaustedException(message);
+
+        await ToolResultAsync(
+            callId: $"budget_{exhaustedCurrency}",
+            result: null,
+            error: new ToolError
+            {
+                Code = ErrorCode.BudgetExhausted,
+                Message = message,
+                Retryable = false,
+            },
+            cancellationToken: cancellationToken).ConfigureAwait(false);
+    }
 
     /// <summary>Artifact ref (asynchronous).</summary>
     public ValueTask ArtifactRefAsync(string uri, string? contentType = null, long? byteSize = null, string? sha256 = null, CancellationToken cancellationToken = default) =>

src/Arcp.Runtime/JobManager.cs

@@ -35,6 +35,7 @@ public sealed partial class JobManager
     private readonly ILoggerFactory _loggers;
     private readonly CredentialManager? _credentials;
     private readonly int _idempotencyWindowSec;
+    private readonly bool _fatalBudgetExhaustion;
 
     /// <summary>Stored record for an idempotency key: original submission fingerprint plus issue time.</summary>
     private sealed record IdempotencyRecord(JobId JobId, string Fingerprint, DateTimeOffset CreatedAt);
@@ -46,14 +47,16 @@ public JobManager(
         TimeProvider time,
         ILoggerFactory loggers,
         CredentialManager? credentials = null,
-        int idempotencyWindowSec = 3600)
+        int idempotencyWindowSec = 3600,
+        bool fatalBudgetExhaustion = false)
     {
         _agents = agents;
         _leases = leases;
         _time = time;
         _loggers = loggers;
         _credentials = credentials;
         _idempotencyWindowSec = idempotencyWindowSec > 0 ? idempotencyWindowSec : 3600;
+        _fatalBudgetExhaustion = fatalBudgetExhaustion;
     }
 
     /// <summary>Initializes a new <see cref="JobManager"/> without credential provisioning.</summary>
@@ -177,7 +180,7 @@ private static JobAcceptedPayload BuildAccepted(Job job)
     public async Task RunAsync(Job job, IAgent agent, Func<Envelope, CancellationToken, ValueTask> emit, CancellationToken cancellationToken)
     {
         job.MarkRunning();
-        var ctx = new JobContext(job, _loggers.CreateLogger($"Arcp.Job.{job.JobId.Value}"), _credentials, _leases);
+        var ctx = new JobContext(job, _loggers.CreateLogger($"Arcp.Job.{job.JobId.Value}"), _credentials, _fatalBudgetExhaustion, _leases);
 
         // Watchdog cancellation source — cancelled in `finally` so the watchdog never outlives
         // the job and never emits a late lease-expired event after the terminal result.

src/Arcp.Runtime/PublicAPI.Unshipped.txt

@@ -70,7 +70,7 @@ Arcp.Runtime.Job.CancellationSource.get -> System.Threading.CancellationTokenSou
 Arcp.Runtime.Job.CancellationToken.get -> System.Threading.CancellationToken
 Arcp.Runtime.Job.CreatedAt.get -> System.DateTimeOffset
 Arcp.Runtime.Job.EmitEventAsync(string! kind, object! body, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.ValueTask
-Arcp.Runtime.Job.EmitMetricAsync(Arcp.Core.Messages.MetricBody! body, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.ValueTask
+Arcp.Runtime.Job.EmitMetricAsync(Arcp.Core.Messages.MetricBody! body, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.ValueTask<string?>
 Arcp.Runtime.Job.IdempotencyKey.get -> string?
 Arcp.Runtime.Job.InlineResultEmitted.get -> bool
 Arcp.Runtime.Job.Input.get -> System.Text.Json.JsonElement?
@@ -197,10 +197,12 @@ Arcp.Runtime.Credentials.InMemoryCredentialStore.RemoveAsync(Arcp.Core.Ids.JobId
 Arcp.Runtime.Job.Credentials.get -> System.Collections.Generic.IReadOnlyList<Arcp.Core.Messages.ProvisionedCredential!>!
 Arcp.Runtime.JobContext.Credentials.get -> System.Collections.Generic.IReadOnlyList<Arcp.Core.Messages.ProvisionedCredential!>!
 Arcp.Runtime.JobContext.RotateCredentialAsync(string! credentialId, Arcp.Core.Messages.ProvisionedCredential! next, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
-Arcp.Runtime.JobManager.JobManager(Arcp.Runtime.Agents.AgentRegistry! agents, Arcp.Runtime.Leases.LeaseManager! leases, System.TimeProvider! time, Microsoft.Extensions.Logging.ILoggerFactory! loggers, Arcp.Runtime.Credentials.CredentialManager? credentials = null, int idempotencyWindowSec = 3600) -> void
+Arcp.Runtime.JobManager.JobManager(Arcp.Runtime.Agents.AgentRegistry! agents, Arcp.Runtime.Leases.LeaseManager! leases, System.TimeProvider! time, Microsoft.Extensions.Logging.ILoggerFactory! loggers, Arcp.Runtime.Credentials.CredentialManager? credentials = null, int idempotencyWindowSec = 3600, bool fatalBudgetExhaustion = false) -> void
 Arcp.Runtime.Job.LeaseExpired.get -> bool
 Arcp.Runtime.Job.MaxRuntimeSec.get -> int?
 Arcp.Runtime.Job.RuntimeLimitExceeded.get -> bool
+Arcp.Runtime.ArcpServerOptions.FatalBudgetExhaustion.get -> bool
+Arcp.Runtime.ArcpServerOptions.FatalBudgetExhaustion.init -> void
 Arcp.Runtime.ArcpServerOptions.IdempotencyWindowSec.get -> int
 Arcp.Runtime.ArcpServerOptions.IdempotencyWindowSec.init -> void
 Arcp.Runtime.JobManager.SubmitAsync(Arcp.Core.Messages.JobSubmitPayload! submit, Arcp.Core.Ids.SessionId sessionId, string? submitterPrincipal, System.Func<Arcp.Core.Wire.Envelope!, System.Threading.CancellationToken, System.Threading.Tasks.ValueTask>! emit, Arcp.Core.Ids.TraceId? inboundTraceId, System.Threading.CancellationToken parentCancellation, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<(Arcp.Runtime.Job! Job, Arcp.Core.Messages.JobAcceptedPayload! Accepted)>!

tests/Arcp.IntegrationTests/BudgetEnforcementTests.cs

@@ -17,16 +17,63 @@ namespace Arcp.IntegrationTests;
 public class BudgetEnforcementTests
 {
     [Fact]
-    public async Task Job_emitting_metric_that_exhausts_cost_budget_terminates_with_BUDGET_EXHAUSTED()
+    public async Task Budget_exhaustion_emits_non_fatal_tool_result_error_and_agent_finishes_normally()
     {
+        // Spec §9.6 SHOULD: prefer surfacing exhaustion as a `tool_result.error` so the agent
+        // may decide whether to continue with non-cost-bearing operations.
         var server = new ArcpServer(new ArcpServerOptions
         {
             Runtime = new RuntimeInfo { Name = "test-runtime", Version = "1.0.0" },
         });
         server.RegisterAgent("spender", async (ctx, ct) =>
         {
             await ctx.MetricAsync("cost.inference", 1.50, unit: "USD", cancellationToken: ct);
-            // Should not reach here — the metric exhausts USD budget.
+            // Agent may continue with non-cost-bearing work and emit a partial result.
+            await ctx.LogAsync("info", "exhausted; returning partial", ct);
+            return "partial";
+        });
+        var (client, srv) = MemoryTransport.Pair();
+        _ = Task.Run(() => server.AcceptAsync(srv));
+
+        await using var c = await ArcpClient.ConnectAsync(client, new ArcpClientOptions
+        {
+            Client = new ClientInfo { Name = "test", Version = "1.0" },
+        });
+
+        var lease = new Lease(new Dictionary<string, IReadOnlyList<string>>
+        {
+            [LeaseNamespaces.CostBudget] = new[] { "USD:1.00" },
+        });
+        var handle = await c.SubmitAsync("spender", leaseRequest: lease);
+
+        // Drain events on the foreground; the channel completes once the terminal envelope arrives.
+        var sawBudgetExhausted = false;
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(3));
+        await foreach (var ev in handle.Events(cts.Token))
+        {
+            if (ev.Kind != "tool_result") continue;
+            var body = ev.BodyAs<ToolResultBody>();
+            if (body?.Error?.Code == ErrorCode.BudgetExhausted) sawBudgetExhausted = true;
+        }
+
+        var result = await handle.Result.WaitAsync(TimeSpan.FromSeconds(3));
+
+        result.Success.Should().BeTrue();
+        sawBudgetExhausted.Should().BeTrue(
+            because: "spec §9.6 SHOULD: exhaustion surfaces as a tool_result.error so the agent can continue");
+    }
+
+    [Fact]
+    public async Task FatalBudgetExhaustion_option_restores_legacy_terminal_BUDGET_EXHAUSTED_behavior()
+    {
+        var server = new ArcpServer(new ArcpServerOptions
+        {
+            Runtime = new RuntimeInfo { Name = "test-runtime", Version = "1.0.0" },
+            FatalBudgetExhaustion = true,
+        });
+        server.RegisterAgent("spender", async (ctx, ct) =>
+        {
+            await ctx.MetricAsync("cost.inference", 1.50, unit: "USD", cancellationToken: ct);
             await Task.Delay(500, ct);
             return "should-not-complete";
         });