fix: idempotency, cancel authority, client robustness, lease boundary (#71, #72, #73, #74, #75, #76)

- #71 idempotent job.submit replay no longer re-runs the agent (skip Resolve/RunAsync on replay)
- #72 lease glob "/prefix/**" keeps the path-boundary separator so siblings are not authorized
- #73 server-rejected job.submit/list_jobs now fault the awaiting client instead of hanging
- #74 cancel authority is session-scoped (fail-closed), not principal-scoped
- #75 add job.cancelled ack and JOB_NOT_FOUND for unknown jobs
- #76 populate session.jobs last_event_seq from a per-job high-water mark

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: nficano

src/Arcp.Client/ArcpClient.Dispatch.cs

@@ -130,15 +130,32 @@ private ValueTask RespondToPingAsync(SessionPingPayload p, CancellationToken can
 
     private void PropagateSessionError(SessionErrorPayload err)
     {
+        var jobError = new JobErrorPayload
+        {
+            Code = err.Code,
+            Message = err.Message,
+            Retryable = err.Retryable,
+            Detail = err.Detail,
+        };
+
         foreach (var h in _handles.Values)
         {
-            h.OnError(new JobErrorPayload
-            {
-                Code = err.Code,
-                Message = err.Message,
-                Retryable = err.Retryable,
-                Detail = err.Detail,
-            });
+            h.OnError(jobError);
+        }
+
+        // A submission rejected before acceptance lives in _pendingSubmits, not _handles, and a
+        // list_jobs request lives in _listJobsRequests. session.error is not correlated to a
+        // specific request id, so the safe contract is to fault every outstanding request — leaving
+        // them pending would hang SubmitAsync/ListJobsAsync until the caller's token fires.
+        while (_pendingSubmits.TryDequeue(out var pending))
+        {
+            pending.OnError(jobError);
+        }
+
+        foreach (var key in _listJobsRequests.Keys)
+        {
+            if (_listJobsRequests.TryRemove(key, out var tcs))
+                tcs.TrySetException(JobHandle.ToException(err.Code, err.Message, err.Detail));
         }
     }
 }

src/Arcp.Client/JobHandle.cs

@@ -81,10 +81,35 @@ internal void OnResult(JobResultPayload payload)
 
     internal void OnError(JobErrorPayload payload)
     {
+        // If the job was rejected before acceptance (e.g. a server session.error for a duplicate
+        // key or unavailable agent), the awaiter on Accepted must fault rather than hang forever.
+        // For a post-acceptance terminal error, Accepted is already resolved so this is a no-op.
+        _accepted.TrySetException(ToException(payload.Code, payload.Message, payload.Detail));
         _terminal.TrySetResult(new JobResult(false, null, payload));
         _events.Writer.TryComplete();
     }
 
+    /// <summary>Map a wire error code to the most specific <see cref="ArcpException"/> subtype so
+    /// callers can <c>catch</c> on the concrete type (e.g. <see cref="DuplicateKeyException"/>).</summary>
+    internal static ArcpException ToException(string code, string message, string? detail) => code switch
+    {
+        ErrorCode.DuplicateKey => new DuplicateKeyException(message, detail),
+        ErrorCode.AgentNotAvailable => new AgentNotAvailableException(message, detail),
+        ErrorCode.AgentVersionNotAvailable => new AgentVersionNotAvailableException(message, detail),
+        ErrorCode.LeaseSubsetViolation => new LeaseSubsetViolationException(message, detail),
+        ErrorCode.PermissionDenied => new PermissionDeniedException(message, detail),
+        ErrorCode.JobNotFound => new JobNotFoundException(message, detail),
+        ErrorCode.InvalidRequest => new InvalidRequestException(message, detail),
+        ErrorCode.Unauthenticated => new UnauthenticatedException(message, detail),
+        ErrorCode.BudgetExhausted => new BudgetExhaustedException(message, detail),
+        ErrorCode.LeaseExpired => new LeaseExpiredException(message, detail),
+        ErrorCode.ResumeWindowExpired => new ResumeWindowExpiredException(message, detail),
+        ErrorCode.HeartbeatLost => new HeartbeatLostException(message, detail),
+        ErrorCode.Timeout => new Arcp.Core.Errors.TimeoutException(message, detail),
+        ErrorCode.Cancelled => new CancelledException(message, detail),
+        _ => new ArcpException(code, message, detail),
+    };
+
     /// <summary>Events.</summary>
     public async IAsyncEnumerable<JobEvent> Events([EnumeratorCancellation] CancellationToken cancellationToken = default)
     {

src/Arcp.Core/Envelope/MessageTypeRegistry.cs

@@ -56,6 +56,7 @@ public static MessageTypeRegistry CreateCoreCatalog()
         r.Register(MessageTypeNames.JobResult, typeof(JobResultPayload));
         r.Register(MessageTypeNames.JobError, typeof(JobErrorPayload));
         r.Register(MessageTypeNames.JobCancel, typeof(JobCancelPayload));
+        r.Register(MessageTypeNames.JobCancelled, typeof(JobCancelledPayload));
         r.Register(MessageTypeNames.JobSubscribe, typeof(JobSubscribePayload));
         r.Register(MessageTypeNames.JobSubscribed, typeof(JobSubscribedPayload));
         r.Register(MessageTypeNames.JobUnsubscribe, typeof(JobUnsubscribePayload));

src/Arcp.Core/Messages/JobCancelledPayload.cs

@@ -0,0 +1,16 @@
+// SPDX-License-Identifier: Apache-2.0
+using System.Text.Json.Serialization;
+
+namespace Arcp.Core.Messages;
+
+/// <summary>Acknowledgement the runtime sends to the submitting session when a <c>job.cancel</c> is
+/// accepted (spec §7.4). The terminal <c>job.error{CANCELLED, final_status:"cancelled"}</c> follows
+/// once the run-loop unwinds the agent.</summary>
+public sealed record JobCancelledPayload
+{
+    /// <summary>Gets the job id.</summary>
+    [JsonPropertyName("job_id")] public required string JobId { get; init; }
+
+    /// <summary>Gets the reason.</summary>
+    [JsonPropertyName("reason")] public string? Reason { get; init; }
+}

src/Arcp.Core/Messages/MessageTypeNames.cs

@@ -51,6 +51,8 @@ public static class MessageTypeNames
     public const string JobError = "job.error";
     /// <summary>Gets the job cancel.</summary>
     public const string JobCancel = "job.cancel";
+    /// <summary>Gets the job cancelled acknowledgement (spec §7.4).</summary>
+    public const string JobCancelled = "job.cancelled";
     /// <summary>Gets the job subscribe.</summary>
     public const string JobSubscribe = "job.subscribe";
     /// <summary>Gets the job subscribed.</summary>
@@ -63,7 +65,7 @@ public static class MessageTypeNames
     {
         SessionHello, SessionWelcome, SessionBye, SessionPing, SessionPong, SessionAck,
         SessionListJobs, SessionJobs, SessionError, SessionResume,
-        JobSubmit, JobAccepted, JobEvent, JobResult, JobError, JobCancel,
+        JobSubmit, JobAccepted, JobEvent, JobResult, JobError, JobCancel, JobCancelled,
         JobSubscribe, JobSubscribed, JobUnsubscribe,
     }.ToFrozenSet();
 }

src/Arcp.Core/PublicAPI.Unshipped.txt

@@ -266,6 +266,19 @@ static Arcp.Core.Messages.ProvisionedCredential.operator ==(Arcp.Core.Messages.P
 Arcp.Core.Messages.JobCancelPayload.JobId.init -> void
 Arcp.Core.Messages.JobCancelPayload.Reason.get -> string?
 Arcp.Core.Messages.JobCancelPayload.Reason.init -> void
+Arcp.Core.Messages.JobCancelledPayload
+Arcp.Core.Messages.JobCancelledPayload.<Clone>$() -> Arcp.Core.Messages.JobCancelledPayload!
+Arcp.Core.Messages.JobCancelledPayload.Equals(Arcp.Core.Messages.JobCancelledPayload? other) -> bool
+Arcp.Core.Messages.JobCancelledPayload.JobCancelledPayload() -> void
+Arcp.Core.Messages.JobCancelledPayload.JobId.get -> string!
+Arcp.Core.Messages.JobCancelledPayload.JobId.init -> void
+Arcp.Core.Messages.JobCancelledPayload.Reason.get -> string?
+Arcp.Core.Messages.JobCancelledPayload.Reason.init -> void
+override Arcp.Core.Messages.JobCancelledPayload.Equals(object? obj) -> bool
+override Arcp.Core.Messages.JobCancelledPayload.GetHashCode() -> int
+override Arcp.Core.Messages.JobCancelledPayload.ToString() -> string!
+static Arcp.Core.Messages.JobCancelledPayload.operator !=(Arcp.Core.Messages.JobCancelledPayload? left, Arcp.Core.Messages.JobCancelledPayload? right) -> bool
+static Arcp.Core.Messages.JobCancelledPayload.operator ==(Arcp.Core.Messages.JobCancelledPayload? left, Arcp.Core.Messages.JobCancelledPayload? right) -> bool
 Arcp.Core.Messages.JobErrorPayload
 Arcp.Core.Messages.JobErrorPayload.<Clone>$() -> Arcp.Core.Messages.JobErrorPayload!
 Arcp.Core.Messages.JobErrorPayload.Code.get -> string!
@@ -707,6 +720,7 @@ const Arcp.Core.Messages.EventKinds.ToolCall = "tool_call" -> string!
 const Arcp.Core.Messages.EventKinds.ToolResult = "tool_result" -> string!
 const Arcp.Core.Messages.MessageTypeNames.JobAccepted = "job.accepted" -> string!
 const Arcp.Core.Messages.MessageTypeNames.JobCancel = "job.cancel" -> string!
+const Arcp.Core.Messages.MessageTypeNames.JobCancelled = "job.cancelled" -> string!
 const Arcp.Core.Messages.MessageTypeNames.JobError = "job.error" -> string!
 const Arcp.Core.Messages.MessageTypeNames.JobEvent = "job.event" -> string!
 const Arcp.Core.Messages.MessageTypeNames.JobResult = "job.result" -> string!

src/Arcp.Runtime/Job.cs

@@ -27,6 +27,7 @@ public sealed class Job
     private readonly object _eventBufferGate = new();
     private readonly List<Envelope> _eventBuffer = [];
     private readonly int _eventBufferCapacity;
+    private long _lastEmittedSeq;
 
     /// <summary>Gets the job id.</summary>
     public JobId JobId { get; }
@@ -192,6 +193,18 @@ public async ValueTask EmitEventAsync(string kind, object body, CancellationToke
         await _emit(env, cancellationToken).ConfigureAwait(false);
     }
 
+    /// <summary>Highest <c>event_seq</c> high-water mark emitted by this job, or <see langword="null"/>
+    /// if it has not emitted any events yet. Surfaced in <c>session.jobs</c> (spec §6.6) so a dashboard
+    /// can decide where to subscribe from.</summary>
+    internal long? LastEmittedSeq
+    {
+        get
+        {
+            var seq = Interlocked.Read(ref _lastEmittedSeq);
+            return seq > 0 ? seq : null;
+        }
+    }
+
     private void BufferEvent(Envelope env)
     {
         // Spec §7.6: bounded per-job history so a later subscriber with `history: true`
@@ -205,6 +218,9 @@ private void BufferEvent(Envelope env)
                 _eventBuffer.RemoveRange(0, _eventBuffer.Count - _eventBufferCapacity);
             }
         }
+
+        // Spec §6.6: track a monotonic per-job high-water mark for last_event_seq in the listing.
+        Interlocked.Increment(ref _lastEmittedSeq);
     }
 
     /// <summary>Snapshot of all events buffered for replay on a new subscription.</summary>

src/Arcp.Runtime/JobManager.Listing.cs

@@ -60,6 +60,7 @@ private static List<Job> ApplyFilter(List<Job> jobs, JobListFilter? filter)
         ParentJobId = j.ParentJobId,
         CreatedAt = j.CreatedAt,
         TraceId = j.TraceId?.Value,
+        LastEventSeq = j.LastEmittedSeq,
     };
 
     private static string MapStatus(JobStatus s) => s switch

src/Arcp.Runtime/JobManager.cs

@@ -85,7 +85,7 @@ public bool TryGet(JobId id, out Job? job)
     /// <summary>Submit a job. The caller (SessionState) hands in the envelope; this method returns
     /// the <see cref="Job"/> to run asynchronously plus the <c>job.accepted</c> payload.
     /// <paramref name="inboundTraceId"/> propagates the envelope's <c>trace_id</c> per spec §11.</summary>
-    public async Task<(Job Job, JobAcceptedPayload Accepted)> SubmitAsync(
+    public async Task<(Job Job, JobAcceptedPayload Accepted, bool IsReplay)> SubmitAsync(
         JobSubmitPayload submit,
         SessionId sessionId,
         string? submitterPrincipal,
@@ -113,7 +113,9 @@ public bool TryGet(JobId id, out Job? job)
                     }
                     if (_jobs.TryGetValue(existingRecord.JobId, out var existing))
                     {
-                        return (existing, BuildAccepted(existing));
+                        // Spec §7.2: an idempotent replay returns the *same* job.accepted and MUST
+                        // NOT re-run the agent. Flag it so the caller skips Resolve/RunAsync.
+                        return (existing, BuildAccepted(existing), true);
                     }
                 }
                 else
@@ -153,7 +155,7 @@ public bool TryGet(JobId id, out Job? job)
             _idempotency[idemKey] = new IdempotencyRecord(jobId, fingerprint, _time.GetUtcNow());
         }
 
-        return (job, BuildAccepted(job));
+        return (job, BuildAccepted(job), false);
     }
 
     private void AssertChildLeaseIsSubset(string parentJobId, Lease child, LeaseConstraints? childConstraints)
@@ -440,15 +442,19 @@ private async Task RunRuntimeWatchdog(Job job, TimeSpan limit, CancellationToken
     private static bool IsTerminal(JobStatus s) =>
         s is JobStatus.Success or JobStatus.Error or JobStatus.Cancelled or JobStatus.TimedOut;
 
-    /// <summary>Cancel a running job. Only the original submitter may cancel; subscribers may not (spec §7.6).</summary>
-    public bool Cancel(JobId jobId, string? requesterPrincipal, string? reason)
+    /// <summary>Cancel a running job. Cancellation authority is scoped to the *submitting session*
+    /// (spec §7.6, §13.3, §14): only the session that submitted the job may cancel it. Subscription —
+    /// even from another session of the same principal — does NOT confer cancel authority. Returns
+    /// <see langword="false"/> if the job does not exist; throws <see cref="PermissionDeniedException"/>
+    /// when the requesting session is not the submitter.</summary>
+    public bool Cancel(JobId jobId, SessionId requesterSession, string? reason)
     {
         if (!_jobs.TryGetValue(jobId, out var job)) return false;
-        // Spec §7.6: subscription does NOT grant cancel authority; only submitter may cancel.
-        if (requesterPrincipal is not null && job.SubmitterPrincipal is not null &&
-            !string.Equals(requesterPrincipal, job.SubmitterPrincipal, StringComparison.Ordinal))
+        // Fail closed: compare the submitting session, never the principal. A null/foreign requester
+        // must not be able to bypass the check (spec §14: "Subscription MUST NOT confer cancel authority").
+        if (!job.SessionId.Equals(requesterSession))
         {
-            throw new PermissionDeniedException("Subscribers MUST NOT cancel jobs (spec §7.6)");
+            throw new PermissionDeniedException("Only the submitting session may cancel a job (spec §7.6, §14)");
         }
         job.CancellationSource.Cancel();
         return true;

src/Arcp.Runtime/Leases/LeaseManager.cs

@@ -85,7 +85,7 @@ public void AssertSubset(Lease parent, Lease child, IReadOnlyDictionary<string,
                 if (childExp > parentExp)
                     throw new LeaseSubsetViolationException("Child lease_constraints.expires_at MUST NOT exceed parent's (spec §9.4)");
             }
-            // No child constraints → child implicitly inherits parent expiry. (spec §9.4)
+            // No child constraints âÿÿ child implicitly inherits parent expiry. (spec §9.4)
         }
     }
 
@@ -153,8 +153,12 @@ internal static bool GlobMatch(string input, string pattern)
         // Convert to regex-ish prefix/suffix.
         if (pattern.EndsWith("/**", StringComparison.Ordinal))
         {
-            var prefix = pattern[..^3];
-            return input.StartsWith(prefix, StringComparison.Ordinal);
+            // Keep the trailing separator so the match respects the path boundary (spec ?9.3):
+            // "/workspace/myapp/**" authorizes "/workspace/myapp" itself and anything strictly
+            // beneath "/workspace/myapp/", but NOT siblings like "/workspace/myapp-private/x".
+            var dir = pattern[..^2];                                   // "/workspace/myapp/"
+            return input.StartsWith(dir, StringComparison.Ordinal)
+                || input.Equals(dir[..^1], StringComparison.Ordinal);  // the directory itself
         }
         if (pattern.EndsWith("*", StringComparison.Ordinal))
         {