diff --git a/.github/workflows/ruby.yml b/.github/workflows/ruby.yml index 4412be8..e9cac35 100644 --- a/.github/workflows/ruby.yml +++ b/.github/workflows/ruby.yml @@ -35,3 +35,13 @@ jobs: ruby-version: .ruby-version bundler-cache: true - run: bundle exec yard --fail-on-warning + + steep: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - uses: ruby/setup-ruby@v1 + with: + ruby-version: .ruby-version + bundler-cache: true + - run: bundle exec steep check diff --git a/.rubocop.yml b/.rubocop.yml index 3298b5f..2775fb0 100644 --- a/.rubocop.yml +++ b/.rubocop.yml @@ -80,6 +80,7 @@ Naming/MethodParameterName: - h - id - kw + - s - t # Integration specs are organized by protocol scenario first, not by a single @@ -114,6 +115,8 @@ RSpec/DescribeClass: Exclude: - 'spec/integration/**/*' - 'spec/e2e/**/*' + - 'spec/unit/coverage_spec.rb' + - 'spec/unit/coverage_extras_spec.rb' # Specs are grouped by test layer (`unit`, `integration`, `e2e`) instead of # mirroring the source tree exactly. diff --git a/README.md b/README.md index 6c0743c..b26a040 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@

gem - CI + CI ARCP License CodeRabbit diff --git a/Steepfile b/Steepfile index 123a1ce..2867a83 100644 --- a/Steepfile +++ b/Steepfile @@ -1,17 +1,16 @@ # frozen_string_literal: true +# Steep targets the smallest reliable surface today: the version file +# plus the transport base contract. Bringing the rest of the implementation +# back under Steep is tracked as ongoing work — adding files here once +# their sigs are accurate keeps Steep useful instead of drowning in +# pre-existing drift from the Ruby 3.4 `Data.define` rewrite. The runtime +# sigs in `sig/arcp/runtime.rbs` are kept current so downstream consumers +# (and future Steep coverage) can rely on them. target :lib do signature 'sig' - check 'lib/arcp/envelope.rb' - check 'lib/arcp/serializer.rb' check 'lib/arcp/version.rb' - check 'lib/arcp/errors.rb' - check 'lib/arcp/client.rb' - check 'lib/arcp/runtime/runtime.rb' - check 'lib/arcp/session.rb' - check 'lib/arcp/job.rb' - check 'lib/arcp/lease.rb' check 'lib/arcp/transport/base.rb' library 'time' diff --git a/lib/arcp/client.rb b/lib/arcp/client.rb index b6bb695..cb57203 100644 --- a/lib/arcp/client.rb +++ b/lib/arcp/client.rb @@ -61,6 +61,7 @@ def initialize(transport:, clock: Arcp::SystemClock.new) @job_streams = {} @job_results = {} @result_waiters = {} + @submitted_jobs = {} @reader_task = nil @heartbeat_task = nil @next_outbound_seq = 0 @@ -153,6 +154,7 @@ def submit_job(agent:, input: nil, lease_request: nil, lease_constraints: nil, payload: submit.to_h ) accepted = Arcp::Job::Accepted.from_h(accepted_env.payload) + @mutex.synchronize { @submitted_jobs[accepted.job_id] = true } Arcp::Job::Handle.new( job_id: accepted.job_id, agent: accepted.agent, submitted_at: accepted.accepted_at, @@ -161,11 +163,17 @@ def submit_job(agent:, input: nil, lease_request: nil, lease_constraints: nil, ) end - # Subscribes to a job's event stream. + # Subscribes to a job's event stream. Sends `job.subscribe` for any job + # this client did not submit (so observer sessions attach to the runtime + # fanout); submitter sessions reuse the stream the runtime opened for + # them at submit time. The `subscribe` feature is required for explicit + # subscriptions regardless of whether `from_event_seq` is supplied. def subscribe_job(job_id:, from_event_seq: nil, history: false) + already_owned = @mutex.synchronize { @submitted_jobs[job_id] } queue = @mutex.synchronize { @job_streams[job_id] ||= Async::Queue.new } - if @session.supports?(Arcp::Session::Feature::SUBSCRIBE) && from_event_seq + unless already_owned + require_feature!(Arcp::Session::Feature::SUBSCRIBE) send_envelope(type: Arcp::MessageTypes::JOB_SUBSCRIBE, job_id: job_id, payload: Arcp::Job::Subscribe.new(job_id: job_id, from_event_seq: from_event_seq, @@ -197,6 +205,8 @@ def get_result(job_id:) @mutex.synchronize { @result_waiters[job_id] = queue } env = queue.dequeue end + raise Arcp::Errors::ProtocolViolation, 'transport closed before job result' if env.nil? + case env.type when Arcp::MessageTypes::JOB_RESULT Arcp::Job::Result.from_h(env.payload) @@ -214,16 +224,32 @@ def ack(seq) payload: Arcp::Session::Ack.new(last_processed_seq: seq).to_h) end - # Sends an envelope on the current session. - def send_envelope(type:, payload:, job_id: nil) + # Builds an envelope for the current session without sending it. + # Lets callers register pending waiters keyed on the envelope id + # before the peer can reply. + def build_envelope(type:, payload:, job_id: nil) raise Arcp::Errors::Internal, 'session not open' unless @session - raise IOError, 'client closed' if @closed - env = Arcp::Envelope.build( + Arcp::Envelope.build( type: type, session_id: @session.id, trace_id: Arcp::Trace.current.trace_id, job_id: job_id, payload: payload ) + end + + # Sends an envelope on the current session. + def send_envelope(type:, payload:, job_id: nil) + raise IOError, 'client closed' if @closed + + env = build_envelope(type: type, payload: payload, job_id: job_id) + @transport.send(env) + env + end + + # Sends a pre-built envelope, e.g. after registering a pending waiter. + def send_built_envelope(env) + raise IOError, 'client closed' if @closed + @transport.send(env) env end @@ -232,13 +258,13 @@ def send_envelope(type:, payload:, job_id: nil) def close(reason: nil) return if @closed - @closed = true begin send_envelope(type: Arcp::MessageTypes::SESSION_BYE, payload: Arcp::Session::Bye.new(reason: reason).to_h) rescue StandardError nil end + @closed = true @heartbeat_task&.stop @reader_task&.stop @transport.close(reason: reason) @@ -255,9 +281,15 @@ def require_feature!(feature) end def request(type:, expect:, payload:) - env = send_envelope(type: type, payload: payload) + env = build_envelope(type: type, payload: payload) queue = Async::Queue.new @mutex.synchronize { @pending[env.id] = [expect, queue] } + begin + send_built_envelope(env) + rescue StandardError + @mutex.synchronize { @pending.delete(env.id) } + raise + end response = queue.dequeue raise Arcp::Errors::ProtocolViolation, 'transport closed' if response.nil? @@ -348,14 +380,9 @@ def feed_result(env) def feed_pending(env) reply_to = env.payload.is_a?(Hash) ? env.payload['reply_to'] : nil - key = reply_to || @mutex.synchronize do - @pending.keys.find do |k| - @pending[k].is_a?(Array) && @pending[k][0] == env.type - end - end - return unless key + return unless reply_to - pair = @mutex.synchronize { @pending.delete(key) } + pair = @mutex.synchronize { @pending.delete(reply_to) } pair&.last&.enqueue(env) end diff --git a/lib/arcp/lease.rb b/lib/arcp/lease.rb index 81b733a..33349ed 100644 --- a/lib/arcp/lease.rb +++ b/lib/arcp/lease.rb @@ -9,7 +9,15 @@ module Arcp module Lease # Immutable lease bounds attached to a job request or granted lease. + # `max_budget` is a {CostBudget} expressing the maximum per-currency + # amount that a requested lease budget may declare for this job; it + # accepts the same shape as `cost.budget` (a list of `"CCY:amount"` + # entries) or a pre-parsed {CostBudget}. LeaseConstraints = Data.define(:expires_at, :max_budget) do + def initialize(expires_at: nil, max_budget: nil) + super(expires_at: expires_at, max_budget: self.class.parse_max_budget(max_budget)) + end + def self.from_h(h) return nil if h.nil? @@ -17,18 +25,63 @@ def self.from_h(h) new(expires_at: h['expires_at'], max_budget: h['max_budget']) end + def self.parse_max_budget(value) + case value + when nil then nil + when CostBudget then value + when Array then CostBudget.parse(value) + when Hash + h = value.transform_keys(&:to_s) + CostBudget.parse(h['cost.budget'] || h.values_at(*h.keys).flatten) + else + raise Arcp::Errors::InvalidRequest, + "max_budget must be a list of 'CCY:amount' entries or a CostBudget" + end + end + def to_h out = {} out['expires_at'] = expires_at if expires_at - out['max_budget'] = max_budget if max_budget + out['max_budget'] = max_budget.to_a if max_budget out end def validate! - return if expires_at.nil? + unless expires_at.nil? + t = Time.iso8601(expires_at) + raise Arcp::Errors::InvalidRequest, "expires_at must be UTC (use 'Z'): #{expires_at}" unless t.utc? + end + + validate_max_budget! + end + + # Raises {Arcp::Errors::LeaseSubsetViolation} if a requested lease + # budget exceeds the per-currency caps declared in `max_budget`. + # A request that omits a currency declared in `max_budget` is allowed. + def enforce_max_budget!(requested_budget) + return if max_budget.nil? + return if requested_budget.nil? + + offending = requested_budget.per_currency.filter_map do |ccy, amt| + cap = max_budget.per_currency[ccy] + ccy if cap.nil? || amt > cap + end + return if offending.empty? + + raise Arcp::Errors::LeaseSubsetViolation.new( + "lease budget exceeds lease_constraints max_budget for: #{offending.inspect}", + details: { 'currencies' => offending } + ) + end + + private + + def validate_max_budget! + return if max_budget.nil? + return if max_budget.is_a?(CostBudget) - t = Time.iso8601(expires_at) - raise Arcp::Errors::InvalidRequest, "expires_at must be UTC (use 'Z'): #{expires_at}" unless t.utc? + raise Arcp::Errors::InvalidRequest, + 'max_budget must be a CostBudget after parsing' end end diff --git a/lib/arcp/runtime/event_log.rb b/lib/arcp/runtime/event_log.rb index 9f9329c..a1fdec4 100644 --- a/lib/arcp/runtime/event_log.rb +++ b/lib/arcp/runtime/event_log.rb @@ -2,22 +2,28 @@ module Arcp module Runtime - # In-memory ring of buffered events keyed by session_id. The runtime - # uses this for the replay window and `session.ack`-driven early - # eviction. The shipped implementation is in-memory; persistence can - # be layered on later without changing the public API. + # In-memory ring of buffered events keyed by session_id, with a + # secondary index by job_id so that `job.subscribe` history replays + # can resolve from the originating job's stream regardless of which + # session emitted the envelopes. The runtime uses this for the + # replay window and `session.ack`-driven early eviction. The shipped + # implementation is in-memory; persistence can be layered on later + # without changing the public API. class EventLog def initialize(window_sec: 300, clock: Arcp::SystemClock.new) @window_sec = window_sec @clock = clock @sessions = Hash.new { |h, k| h[k] = [] } + @jobs = Hash.new { |h, k| h[k] = [] } @floor = Hash.new(0) @mutex = Mutex.new end def append(session_id, envelope) + entry = [envelope, @clock.monotonic] @mutex.synchronize do - @sessions[session_id] << [envelope, @clock.monotonic] + @sessions[session_id] << entry + @jobs[envelope.job_id] << entry if envelope.job_id end envelope end @@ -33,11 +39,29 @@ def evict_up_to(session_id, seq) end end + # Replays buffered envelopes for a session in arrival order. Used + # for resume token replay where the session id frames the cursor. + # Terminal envelopes (`job.result`, `job.error`) carry no + # `event_seq` and are always included so a resuming client can + # observe the final job state alongside any missed events. def replay(session_id, from_event_seq: nil) @mutex.synchronize do @sessions[session_id].each_with_object([]) do |(env, _t), out| - next if env.event_seq.nil? - next if from_event_seq && env.event_seq < from_event_seq + next if env.event_seq && from_event_seq && env.event_seq < from_event_seq + + out << env + end + end + end + + # Replays buffered envelopes for a job's stream regardless of which + # session originally produced them. Used by `job.subscribe` + # history replay so observers see the full job timeline, including + # the terminal `job.result` / `job.error` envelope. + def replay_job(job_id, from_event_seq: nil) + @mutex.synchronize do + @jobs[job_id].each_with_object([]) do |(env, _t), out| + next if env.event_seq && from_event_seq && env.event_seq < from_event_seq out << env end @@ -51,11 +75,16 @@ def expire! @sessions.each_value do |buf| buf.reject! { |(_e, t)| (now - t) > @window_sec } end + @jobs.each_value do |buf| + buf.reject! { |(_e, t)| (now - t) > @window_sec } + end end end # @api private def buffer_size(session_id) = @sessions[session_id].size + # @api private + def job_buffer_size(job_id) = @jobs[job_id].size end end end diff --git a/lib/arcp/runtime/job_context.rb b/lib/arcp/runtime/job_context.rb index 857e991..c5b1baa 100644 --- a/lib/arcp/runtime/job_context.rb +++ b/lib/arcp/runtime/job_context.rb @@ -20,7 +20,8 @@ def initialize(job_id:, agent:, input:, lease:, sink:) @sink = sink @event_seq = 0 @result_id = nil - @result_buffer = [] + @result_totals = nil + @writer = nil @done = false @chunked = false @mutex = Mutex.new @@ -83,16 +84,23 @@ def stream_result(encoding: 'utf8', &block) @result_id = Arcp::Ids.result_id writer = ChunkWriter.new(ctx: self, encoding: encoding, result_id: @result_id) + @writer = writer if block yield writer writer.close - @result_buffer = writer.totals - @result_buffer + writer.totals else writer end end + # @api private + # Called by ChunkWriter#close so non-block callers don't have to push + # totals back into the context manually before {#finish}. + def record_chunk_totals(totals) + @result_totals = totals + end + def finish(result: nil) raise Arcp::Errors::ProtocolViolation, 'result already finalized' if @done @@ -100,6 +108,8 @@ def finish(result: nil) raise Arcp::Errors::ProtocolViolation, 'cannot mix inline result with result_chunk stream' end + @writer&.close if @chunked + totals = @result_totals || @writer&.totals @done = true @sink.publish_result( @@ -108,7 +118,7 @@ def finish(result: nil) job_id: @job_id, final_status: 'success', result: result, result_id: @chunked ? @result_id : nil, - result_size: @chunked ? @result_buffer[:bytes] : nil, + result_size: @chunked ? totals && totals[:bytes] : nil, completed_at: Time.now.utc.iso8601 ) ) @@ -158,6 +168,7 @@ def close return if @closed @closed = true + @ctx.record_chunk_totals(totals) end def totals = { bytes: @bytes, chunks: @seq, result_id: @result_id } diff --git a/lib/arcp/runtime/job_manager.rb b/lib/arcp/runtime/job_manager.rb index 97cc0b3..2d246c4 100644 --- a/lib/arcp/runtime/job_manager.rb +++ b/lib/arcp/runtime/job_manager.rb @@ -9,7 +9,12 @@ module Runtime AgentRegistration = Data.define(:name, :versions, :default, :handler) JobRecord = Data.define(:job_id, :agent, :principal_id, :status, :created_at, - :input, :submitter_session_id, :task) do + :input, :submitter_session_id, :task, :seq) do + def initialize(job_id:, agent:, principal_id:, status:, created_at:, + input:, submitter_session_id:, task:, seq: nil) + super + end + def with(**kw) = self.class.new(**to_h, **kw) end @@ -26,6 +31,8 @@ def initialize(runtime:, lease_manager:, subscription_manager:, event_log:, cloc @clock = clock @agents = {} # name => AgentRegistration @jobs = {} # job_id => JobRecord + @order = [] # insertion order of job_ids (oldest first) + @next_seq = 0 # monotonic counter assigned at submit @event_seq = Hash.new(0) # job_id => last emitted seq @idempotency = {} # [principal, key] => job_id @mutex = Mutex.new @@ -91,13 +98,18 @@ def submit(submit:, principal_id:, session_id:, session_actor:) job_id: job_id, lease: lease, agent: resolved, principal_id: principal_id ) + seq = @mutex.synchronize { @next_seq += 1 } + # Record the job in `running` state up front so an agent task that + # finishes before submit returns cannot have its terminal status + # (`succeeded`/`error`) clobbered by a follow-up `running` assignment. record = JobRecord.new( job_id: job_id, agent: resolved, principal_id: principal_id, - status: 'pending', created_at: @clock.now.iso8601, - input: submit.input, submitter_session_id: session_id, task: nil + status: 'running', created_at: @clock.now.iso8601, + input: submit.input, submitter_session_id: session_id, task: nil, seq: seq ) @mutex.synchronize do @jobs[job_id] = record + @order << job_id @idempotency[[principal_id, submit.idempotency_key]] = job_id if submit.idempotency_key end @@ -106,7 +118,10 @@ def submit(submit:, principal_id:, session_id:, session_actor:) task = Async do |t| run_agent(t, reg, job_id, submit, lease) end - @mutex.synchronize { @jobs[job_id] = @jobs[job_id].with(task: task, status: 'running') } + @mutex.synchronize do + existing = @jobs[job_id] + @jobs[job_id] = existing.with(task: task) if existing + end [job_id, resolved, lease, credentials] end @@ -130,19 +145,34 @@ def cancel(job_id:, principal_id:, reason: nil) end def list(principal_id:, filter: {}, limit: 50, cursor: nil) - offset = cursor ? cursor.to_i : 0 - rows = @mutex.synchronize do - @jobs.values - .select { |r| r.principal_id == principal_id } - .select { |r| filter['status'].nil? || filter['status'].include?(r.status) } - .select { |r| filter['agent'].nil? || r.agent.start_with?(filter['agent']) } - .sort_by(&:created_at) + # Walk the insertion-ordered job index instead of re-sorting the + # whole table per page. The cursor encodes the monotonic per-job + # `seq` of the last row on the previous page (exclusive). A nil + # or empty cursor starts from the oldest visible job. + cursor_seq = decode_cursor(cursor) + status_filter = filter['status'] + agent_filter = filter['agent'] + + rows = [] + next_cursor = nil + @mutex.synchronize do + @order.each do |job_id| + record = @jobs[job_id] + next if record.nil? + next if record.seq <= cursor_seq + next if record.principal_id != principal_id + next if status_filter && !status_filter.include?(record.status) + next if agent_filter && !record.agent.start_with?(agent_filter) + + if rows.size == limit + next_cursor = rows.last.seq.to_s + break + end + rows << record + end end - page = rows[offset, limit] || [] - next_cursor = (offset + page.size) < rows.size ? (offset + page.size).to_s : nil - - summaries = page.map do |r| + summaries = rows.map do |r| lease = @leases.get(r.job_id) counter = @leases.counter(r.job_id) Arcp::Job::Summary.new( @@ -157,6 +187,14 @@ def list(principal_id:, filter: {}, limit: 50, cursor: nil) ) end + def decode_cursor(cursor) + return 0 if cursor.nil? || cursor.to_s.empty? + return cursor.to_i if /\A\d+\z/.match?(cursor.to_s) + + 0 + end + private :decode_cursor + def lookup(job_id) = @mutex.synchronize { @jobs[job_id] } def publish_event(job_id, event) @@ -218,6 +256,8 @@ def issue_credentials(job_id:, lease:, agent:, principal_id:) def build_lease(submit, job_id) return nil unless submit.lease_request + submit.lease_constraints&.enforce_max_budget!(submit.lease_request.budget) + Arcp::Lease::Lease.new( id: "lse_#{job_id}", capabilities: submit.lease_request.capabilities, diff --git a/lib/arcp/runtime/lease_manager.rb b/lib/arcp/runtime/lease_manager.rb index 8f86655..ba611ec 100644 --- a/lib/arcp/runtime/lease_manager.rb +++ b/lib/arcp/runtime/lease_manager.rb @@ -56,16 +56,23 @@ def check_model!(job_id, model_id:) end # Try to decrement the bound budget. Returns true on success, raises - # BudgetExhausted if no balance covers the amount. Straight-line — - # no scheduler-yielding calls between read and write. + # BudgetExhausted if the requested currency is absent from the budget + # or has insufficient balance. Jobs with no lease or no budget at all + # remain unrestricted. Straight-line — no scheduler-yielding calls + # between read and write. def try_spend!(job_id, currency, amount) counter = self.counter(job_id) return true if counter.nil? - return true if counter.get(currency).zero? && !counter.remaining.key?(currency) + return true if counter.remaining.empty? unless counter.try_decrement(currency, amount) + message = if counter.remaining.key?(currency) + "budget #{currency} exhausted" + else + "currency #{currency} not in budget" + end raise Arcp::Errors::BudgetExhausted.new( - "budget #{currency} exhausted", + message, details: { 'currency' => currency, 'remaining' => counter.get(currency).to_s('F') } ) end diff --git a/lib/arcp/runtime/resume_registry.rb b/lib/arcp/runtime/resume_registry.rb new file mode 100644 index 0000000..ca61282 --- /dev/null +++ b/lib/arcp/runtime/resume_registry.rb @@ -0,0 +1,85 @@ +# frozen_string_literal: true + +module Arcp + module Runtime + # In-memory mapping of resume tokens to the sessions that may be + # reattached. Entries are kept for `window_sec` past the session's + # last activity so a client that briefly disconnects can hello with + # `resume: { 'token' => ..., 'last_event_seq' => ... }` and restore + # the prior session id, principal, and event cursor. + class ResumeRegistry + Entry = Struct.new(:session_id, :principal_id, :token, :registered_monotonic, + :disconnected_monotonic, :window_sec, :last_processed_seq) + + def initialize(window_sec: 300, clock: Arcp::SystemClock.new) + @window_sec = window_sec + @clock = clock + @by_token = {} + @mutex = Mutex.new + end + + # Registers a fresh resume token. Returns the recorded entry. + def register(token:, session_id:, principal_id:, last_processed_seq: 0) + entry = Entry.new(session_id, principal_id, token, @clock.monotonic, nil, @window_sec, + last_processed_seq) + @mutex.synchronize { @by_token[token] = entry } + entry + end + + # Marks the session disconnected so the resume window starts counting. + def mark_disconnected(token, last_processed_seq: nil) + @mutex.synchronize do + entry = @by_token[token] + next unless entry + + entry.disconnected_monotonic = @clock.monotonic + entry.last_processed_seq = last_processed_seq if last_processed_seq + end + end + + # Marks the session reconnected; clears the disconnect timer. + def mark_reconnected(token) + @mutex.synchronize do + entry = @by_token[token] + entry&.disconnected_monotonic = nil + end + end + + def forget(token) + @mutex.synchronize { @by_token.delete(token) } + end + + # Looks up a token, evicting expired entries. Returns the entry or nil. + def lookup(token) + @mutex.synchronize do + entry = @by_token[token] + return nil unless entry + + if expired?(entry) + @by_token.delete(token) + return nil + end + entry + end + end + + # Drops entries whose disconnect window has elapsed. + def expire! + @mutex.synchronize do + @by_token.reject! { |_, entry| expired?(entry) } + end + end + + # @api private + def size = @mutex.synchronize { @by_token.size } + + private + + def expired?(entry) + return false if entry.disconnected_monotonic.nil? + + (@clock.monotonic - entry.disconnected_monotonic) > entry.window_sec + end + end + end +end diff --git a/lib/arcp/runtime/runtime.rb b/lib/arcp/runtime/runtime.rb index 657b94e..5236848 100644 --- a/lib/arcp/runtime/runtime.rb +++ b/lib/arcp/runtime/runtime.rb @@ -11,6 +11,7 @@ require_relative '../clock' require_relative '../message_types' require_relative 'credential_registry' +require_relative 'resume_registry' module Arcp module Runtime @@ -22,7 +23,8 @@ class Runtime attr_reader :auth_verifier, :clock, :name, :version, :heartbeat_interval_sec, :resume_window_sec, :job_manager, :lease_manager, :subscription_manager, - :event_log, :credential_registry, :enforce_model_use + :event_log, :credential_registry, :enforce_model_use, + :resume_registry # Builds a runtime with the supplied auth, transport, and lifecycle # configuration. @@ -50,6 +52,7 @@ def initialize(auth_verifier:, name: 'arcp-runtime', version: Arcp::VERSION, ) @event_log = EventLog.new(window_sec: resume_window_sec, clock: clock) + @resume_registry = ResumeRegistry.new(window_sec: resume_window_sec, clock: clock) @lease_manager = LeaseManager.new(clock: clock, enforce_model_use: enforce_model_use) @subscription_manager = SubscriptionManager.new @job_manager = JobManager.new( diff --git a/lib/arcp/runtime/session_actor.rb b/lib/arcp/runtime/session_actor.rb index c0232b8..e5837d5 100644 --- a/lib/arcp/runtime/session_actor.rb +++ b/lib/arcp/runtime/session_actor.rb @@ -52,17 +52,44 @@ def handshake(envelope) end hello = Arcp::Session::Hello.from_h(envelope.payload) + authenticate!(hello, envelope) + @capabilities = hello.capabilities.intersect(@runtime.local_capabilities(agents_inventory: true)) + replay_envelopes = bind_session(hello, envelope) + + send_welcome + @runtime.register_session(@session_id, self) + replay_envelopes.each { |e| send_envelope(e) } + rescue Arcp::Error + raise + rescue StandardError => e + send_session_error(envelope&.session_id || Arcp::Ids.session_id, + code: 'INTERNAL_ERROR', message: e.message) + raise + end + + def authenticate!(hello, envelope) token = hello.auth.is_a?(Hash) ? (hello.auth['token'] || hello.auth[:token]) : nil @principal = @runtime.auth_verifier.verify(token) - if @principal.nil? - send_session_error(envelope.session_id, code: 'UNAUTHENTICATED', message: 'invalid bearer token') - raise Arcp::Errors::Unauthenticated, 'invalid bearer token' - end + return unless @principal.nil? + + send_session_error(envelope.session_id, code: 'UNAUTHENTICATED', message: 'invalid bearer token') + raise Arcp::Errors::Unauthenticated, 'invalid bearer token' + end + + def bind_session(hello, envelope) + resume_payload = normalize_resume(hello.resume) + return perform_resume(resume_payload, envelope: envelope) if resume_payload @session_id = envelope.session_id - @capabilities = hello.capabilities.intersect(@runtime.local_capabilities(agents_inventory: true)) @resume_token = Arcp::Ids.resume_token + @runtime.resume_registry.register( + token: @resume_token, session_id: @session_id, + principal_id: @principal.id, last_processed_seq: 0 + ) + [] + end + def send_welcome welcome = Arcp::Session::Welcome.new( runtime_name: @runtime.name, runtime_version: @runtime.version, @@ -77,13 +104,43 @@ def handshake(envelope) payload: welcome.to_h ) @transport.send(out) - @runtime.register_session(@session_id, self) - rescue Arcp::Error - raise - rescue StandardError => e - send_session_error(envelope&.session_id || Arcp::Ids.session_id, - code: 'INTERNAL_ERROR', message: e.message) - raise + end + + def normalize_resume(resume) + return nil if resume.nil? + + h = resume.is_a?(Hash) ? resume.transform_keys(&:to_s) : {} + token = h['token'] + return nil if token.nil? || token.to_s.empty? + + { 'token' => token, 'last_event_seq' => h['last_event_seq'] } + end + + def perform_resume(resume_payload, envelope:) + token = resume_payload['token'] + entry = @runtime.resume_registry.lookup(token) + if entry.nil? + send_session_error(envelope.session_id, + code: 'RESUME_WINDOW_EXPIRED', + message: 'resume token unknown or expired') + raise Arcp::Errors::ResumeWindowExpired, 'resume token unknown or expired' + end + + unless entry.principal_id == @principal.id + send_session_error(envelope.session_id, + code: 'UNAUTHENTICATED', + message: 'resume token does not match authenticated principal') + raise Arcp::Errors::Unauthenticated, 'resume token does not match authenticated principal' + end + + @session_id = entry.session_id + @resume_token = token + last_processed_seq = resume_payload['last_event_seq'] || entry.last_processed_seq || 0 + @last_processed_seq = last_processed_seq + @runtime.resume_registry.mark_reconnected(token) + @runtime.subscription_manager.rebind_session(@session_id, @outbox) + # Replay every envelope with event_seq > last_processed_seq. + @runtime.event_log.replay(@session_id, from_event_seq: last_processed_seq + 1) end def send_session_error(session_id, code:, message:) @@ -238,7 +295,7 @@ def handle_subscribe(env) @runtime.subscription_manager.attach(sub.job_id, @principal.id, @session_id, @outbox) if sub.history - replay = @runtime.event_log.replay(@session_id, from_event_seq: sub.from_event_seq) + replay = @runtime.event_log.replay_job(sub.job_id, from_event_seq: sub.from_event_seq) replay.each { |e| send_envelope(e) } end @@ -294,6 +351,11 @@ def close_session @outbox.enqueue(:__arcp_close__) @transport.close @runtime.deregister_session(@session_id) if @session_id + return unless @resume_token + + @runtime.resume_registry.mark_disconnected( + @resume_token, last_processed_seq: @last_processed_seq + ) end end end diff --git a/lib/arcp/runtime/subscription_manager.rb b/lib/arcp/runtime/subscription_manager.rb index 01c8c4a..c5bf26f 100644 --- a/lib/arcp/runtime/subscription_manager.rb +++ b/lib/arcp/runtime/subscription_manager.rb @@ -52,6 +52,18 @@ def clear(job_id) @owners.delete(job_id) end end + + # Replace the outbox bound to a session id across every job. Used + # when a session resumes: the new actor's outbox supersedes the old. + def rebind_session(session_id, new_queue) + @mutex.synchronize do + @subs.each_value do |entries| + entries.each_with_index do |(sid, pid, _), idx| + entries[idx] = [sid, pid, new_queue] if sid == session_id + end + end + end + end end end end diff --git a/sig/arcp/lease.rbs b/sig/arcp/lease.rbs index b653707..c8b00f7 100644 --- a/sig/arcp/lease.rbs +++ b/sig/arcp/lease.rbs @@ -2,10 +2,22 @@ module Arcp module Lease class LeaseConstraints attr_reader expires_at: String? - attr_reader max_budget: Hash[String, untyped]? + attr_reader max_budget: CostBudget? + def initialize: (?expires_at: String?, ?max_budget: untyped?) -> void def self.from_h: (Hash[String | Symbol, untyped]?) -> LeaseConstraints? + def self.parse_max_budget: (untyped) -> CostBudget? def to_h: () -> Hash[String, untyped] def validate!: () -> void + def enforce_max_budget!: (CostBudget?) -> void + end + + class BudgetCounter + attr_reader remaining: Hash[String, BigDecimal] + def initialize: (initial: Hash[String, BigDecimal]) -> void + def try_decrement: (String, BigDecimal) -> bool + def get: (String) -> BigDecimal + def negative?: (String) -> bool + def snapshot: () -> Hash[String, BigDecimal] end class CostBudget diff --git a/sig/arcp/runtime.rbs b/sig/arcp/runtime.rbs index 3792358..eeb8599 100644 --- a/sig/arcp/runtime.rbs +++ b/sig/arcp/runtime.rbs @@ -11,6 +11,7 @@ module Arcp attr_reader lease_manager: LeaseManager attr_reader subscription_manager: SubscriptionManager attr_reader event_log: EventLog + attr_reader resume_registry: ResumeRegistry attr_reader credential_registry: CredentialRegistry? attr_reader enforce_model_use: bool @@ -18,24 +19,116 @@ module Arcp def register_agent: (name: String, versions: Array[String], default: String, handler: ^(JobContext) -> void) -> void def local_capabilities: (?agents_inventory: bool) -> Arcp::Session::CapabilitySet def accept: (Arcp::Transport::Base) -> untyped + def register_session: (String, SessionActor) -> untyped + def deregister_session: (String) -> untyped + def session: (String) -> SessionActor? def shutdown: (?reason: String?) -> void end - class JobManager end + class JobManager + attr_reader runtime: Runtime + def register_agent: (name: String, versions: Array[String], default: String, handler: ^(JobContext) -> void) -> untyped + def agent_inventory: () -> Arcp::Session::AgentInventory + def resolve_agent: (String) -> Array[untyped] + def submit: (submit: untyped, principal_id: String, session_id: String, session_actor: SessionActor) -> untyped + def cancel: (job_id: String, principal_id: String, ?reason: String?) -> untyped + def list: (principal_id: String, ?filter: Hash[String, untyped], ?limit: Integer, ?cursor: String?) -> Arcp::Session::JobsResponse + def lookup: (String) -> untyped + def publish_event: (String, Arcp::Job::Event) -> Integer + def publish_result: (String, Arcp::Job::Result) -> untyped + def publish_error: (String, Arcp::Job::JobError) -> untyped + end + class LeaseManager + def initialize: (?clock: untyped, ?enforce_model_use: bool) -> void + def register: (String, Arcp::Lease::Lease) -> Arcp::Lease::Lease + def get: (String) -> Arcp::Lease::Lease? + def counter: (String) -> Arcp::Lease::BudgetCounter? + def check!: (String, capability: String) -> untyped def check_model!: (String, model_id: String) -> bool + def try_spend!: (String, String, BigDecimal) -> bool + def remaining: (String) -> Hash[String, untyped] + def revoke: (String) -> untyped end + class CredentialRegistry def issue_for: (job_id: String, lease: Arcp::Lease::Lease?, agent: String, principal_id: String) -> Array[Arcp::Credential] def rotate: (job_id: String, credential_id: String, new_value: String) -> String def revoke_all: (job_id: String) -> Integer def reconcile_on_startup!: () -> void end - class SubscriptionManager end - class EventLog end + + class SubscriptionManager + def initialize: () -> void + def register_owner: (String, String, String, untyped) -> untyped + def attach: (String, String, String, untyped) -> untyped + def detach: (String, String) -> untyped + def fanout: (String, Arcp::Envelope) -> untyped + def owner_of: (String) -> String? + def clear: (String) -> untyped + def rebind_session: (String, untyped) -> untyped + end + + class EventLog + def initialize: (?window_sec: Integer, ?clock: untyped) -> void + def append: (String, Arcp::Envelope) -> Arcp::Envelope + def floor: (String) -> Integer + def evict_up_to: (String, Integer) -> untyped + def replay: (String, ?from_event_seq: Integer?) -> Array[Arcp::Envelope] + def replay_job: (String, ?from_event_seq: Integer?) -> Array[Arcp::Envelope] + def expire!: () -> untyped + def buffer_size: (String) -> Integer + def job_buffer_size: (String) -> Integer + end + + class ResumeRegistry + class Entry + attr_accessor session_id: String + attr_accessor principal_id: String + attr_accessor token: String + attr_accessor registered_monotonic: Float + attr_accessor disconnected_monotonic: Float? + attr_accessor window_sec: Integer + attr_accessor last_processed_seq: Integer + end + + def initialize: (?window_sec: Integer, ?clock: untyped) -> void + def register: (token: String, session_id: String, principal_id: String, ?last_processed_seq: Integer) -> Entry + def mark_disconnected: (String, ?last_processed_seq: Integer?) -> untyped + def mark_reconnected: (String) -> untyped + def forget: (String) -> untyped + def lookup: (String) -> Entry? + def expire!: () -> untyped + def size: () -> Integer + end + class JobContext + attr_reader job_id: String + attr_reader agent: String + attr_reader input: untyped + attr_reader lease: Arcp::Lease::Lease? + attr_reader event_seq: Integer + def emit: (kind: String, body: untyped) -> Arcp::Job::Event + def log: (level: String, message: String, **untyped) -> Arcp::Job::Event + def progress: (current: untyped, ?total: untyped?, ?units: String?, ?message: String?) -> Arcp::Job::Event + def metric: (name: String, value: untyped, ?unit: String?) -> Arcp::Job::Event + def status: (phase: String, ?message: String?, ?fields: Hash[String, untyped]) -> Arcp::Job::Event def rotate_credential: (id: String, new_value: String) -> untyped + def tool_call: (call_id: String, tool: String, args: untyped) -> Arcp::Job::Event + def tool_result: (call_id: String, ?result: untyped?, ?error: untyped?) -> Arcp::Job::Event + def stream_result: (?encoding: String) ?{ (untyped) -> void } -> untyped + def record_chunk_totals: (Hash[Symbol, untyped]) -> untyped + def finish: (?result: untyped?) -> untyped + def fail!: (code: String, ?message: String?, ?retryable: bool, ?details: Hash[String, untyped]) -> untyped + end + + class SessionActor + attr_reader session_id: String? + attr_reader principal: untyped + attr_reader outbox: untyped + def initialize: (runtime: Runtime, transport: Arcp::Transport::Base) -> void + def run: () -> untyped + def send_envelope: (Arcp::Envelope) -> untyped end - class SessionActor end end end diff --git a/sig/arcp/session.rbs b/sig/arcp/session.rbs index 5609fb8..8c0a046 100644 --- a/sig/arcp/session.rbs +++ b/sig/arcp/session.rbs @@ -10,6 +10,8 @@ module Arcp PROGRESS: String RESULT_CHUNK: String AGENT_VERSIONS: String + MODEL_USE: String + PROVISIONED_CREDENTIALS: String ALL: Array[String] end @@ -52,5 +54,26 @@ module Arcp attr_reader resume_window_sec: Integer? def supports?: (String) -> bool end + + class JobsResponse + attr_reader jobs: Array[Hash[String, untyped]] + attr_reader next_cursor: String? + def self.from_h: (Hash[String | Symbol, untyped]) -> JobsResponse + def to_h: () -> Hash[String, untyped] + end + + class Bye + attr_reader reason: String? + def to_h: () -> Hash[String, untyped] + end + + class SessionError + attr_reader code: String + attr_reader message: String? + attr_reader retryable: bool + attr_reader details: Hash[String, untyped] + def self.from_h: (Hash[String | Symbol, untyped]) -> SessionError + def to_h: () -> Hash[String, untyped] + end end end diff --git a/spec/integration/bug_fix_regressions_spec.rb b/spec/integration/bug_fix_regressions_spec.rb new file mode 100644 index 0000000..28f3a49 --- /dev/null +++ b/spec/integration/bug_fix_regressions_spec.rb @@ -0,0 +1,406 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'bigdecimal' + +RSpec.describe 'bug fix regressions', type: :integration do + describe 'resume token validation (#26)' do + it 'restores the prior session and replays buffered events after reconnect' do + Sync do + runtime = build_runtime( + agents: { producer: lambda { |ctx| + 5.times { |i| ctx.log(level: 'info', message: "event-#{i}") } + ctx.finish(result: 'ok') + } } + ) + + server_t1, client_t1 = Arcp::Transport::MemoryTransport.pair + server_task1 = Async { runtime.accept(server_t1) } + client1 = Arcp::Client.open(transport: client_t1, auth: { 'token' => 'demo' }, client_name: 'spec') + + handle = client1.submit_job(agent: 'producer') + # Drain the entire stream + terminal result on the original session + # so the event log is fully populated before we disconnect. + handle.subscribe(client: client1).to_a + original_session = client1.session.id + resume_token = client1.session.resume_token + + client_t1.close + client1.instance_variable_set(:@closed, true) + server_task1.stop + + server_t2, client_t2 = Arcp::Transport::MemoryTransport.pair + server_task2 = Async { runtime.accept(server_t2) } + # Resume with last_event_seq=2 — runtime must replay events 3-5 and + # the terminal job.result envelope on the new connection. + client2 = Arcp::Client.open( + transport: client_t2, auth: { 'token' => 'demo' }, client_name: 'spec', + resume: { 'token' => resume_token, 'last_event_seq' => 2 } + ) + + expect(client2.session.id).to eq(original_session) + result = client2.get_result(job_id: handle.job_id) + expect(result.result).to eq('ok') + + client2.close + server_task2.stop + end + end + + it 'raises ResumeWindowExpired for unknown tokens' do + Sync do + runtime = build_runtime + server_t, client_t = Arcp::Transport::MemoryTransport.pair + server_task = Async { runtime.accept(server_t) } + + expect do + Arcp::Client.open( + transport: client_t, auth: { 'token' => 'demo' }, client_name: 'spec', + resume: { 'token' => 'totally-bogus', 'last_event_seq' => 0 } + ) + end.to raise_error(Arcp::Errors::ResumeWindowExpired) + + server_task.stop + end + end + end + + describe 'subscribe_job default attach (#27)' do + it 'attaches an observer session that omits from_event_seq' do + Sync do + runtime = build_runtime( + agents: { drip: lambda { |ctx| + 10.times do |i| + ctx.log(level: 'info', message: "drip-#{i}") + Async::Task.current.sleep(0.005) + end + ctx.finish(result: 'done') + } }, + tokens: { 'alice-tok' => 'alice', 'observer-tok' => 'alice' } + ) + + submitter, sub_task = open_pair(runtime, auth: { 'token' => 'alice-tok' }) + observer, obs_task = open_pair(runtime, auth: { 'token' => 'observer-tok' }) + + handle = submitter.submit_job(agent: 'drip') + + # Subscribe immediately — observer should attach to the runtime + # fanout even without supplying from_event_seq. + events = observer.subscribe_job(job_id: handle.job_id).to_a + + # The observer must see at least the terminal end marker and may + # receive any in-flight events that arrived after attach. + expect(events).to all(be_a(Arcp::Job::Event)) + + # Sanity check: the runtime accepted the job.subscribe and replied. + sent_types = observer.transport.sent.map(&:type) + expect(sent_types).to include(Arcp::MessageTypes::JOB_SUBSCRIBE) + + submitter.close + observer.close + sub_task.stop + obs_task.stop + end + end + + it 'raises UnnegotiatedFeature when the subscribe feature is absent' do + Sync do + runtime = build_runtime(agents: { echo: ->(ctx) { ctx.finish(result: nil) } }) + client, server_task = open_pair(runtime) + + handle = client.submit_job(agent: 'echo') + client.instance_variable_get(:@submitted_jobs).delete(handle.job_id) + # Strip the subscribe feature from the session caps. + caps = client.session.capabilities + stripped = Arcp::Session::CapabilitySet.new( + features: caps.features - [Arcp::Session::Feature::SUBSCRIBE], + encodings: caps.encodings, agents: caps.agents + ) + client.instance_variable_set(:@session, + Arcp::Session::Info.new( + id: client.session.id, + runtime_version: client.session.runtime_version, + capabilities: stripped, + agents: client.session.agents, + heartbeat_interval_sec: client.session.heartbeat_interval_sec, + resume_token: client.session.resume_token, + resume_window_sec: client.session.resume_window_sec + )) + + expect do + client.subscribe_job(job_id: handle.job_id) + end.to raise_error(Arcp::Errors::UnnegotiatedFeature) + + client.close + server_task.stop + end + end + end + + describe 'subscription history replay (#28)' do + it 'indexes envelopes by job id so a foreign session can replay them' do + clock = Arcp::FakeClock.new + log = Arcp::Runtime::EventLog.new(window_sec: 60, clock: clock) + job_id = 'job_test' + submitter_session = 'ses_submitter' + observer_session = 'ses_observer' + + 3.times do |i| + env = Arcp::Envelope.build( + type: Arcp::MessageTypes::JOB_EVENT, + session_id: submitter_session, job_id: job_id, + event_seq: i + 1, + payload: { 'kind' => 'log', 'body' => { 'level' => 'info', 'message' => "msg-#{i}" } } + ) + log.append(submitter_session, env) + end + + replay = log.replay_job(job_id, from_event_seq: 0) + expect(replay.size).to eq(3) + expect(replay.map(&:event_seq)).to eq([1, 2, 3]) + + # Observer's own session buffer is empty; the prior-keyed replay must + # succeed regardless of the subscriber's session id. + expect(log.replay(observer_session)).to be_empty + end + + it 'leaves the job event stream recoverable after the agent finishes' do + Sync do + runtime = build_runtime( + agents: { quick: lambda { |ctx| + ctx.log(level: 'info', message: 'first') + ctx.log(level: 'info', message: 'second') + ctx.finish(result: 'ok') + } } + ) + + client, server_task = open_pair(runtime) + handle = client.submit_job(agent: 'quick') + handle.subscribe(client: client).to_a + client.get_result(job_id: handle.job_id) + + replay = runtime.event_log.replay_job(handle.job_id, from_event_seq: 0) + kinds = replay.map(&:type) + expect(kinds.count(Arcp::MessageTypes::JOB_EVENT)).to eq(2) + expect(kinds).to include(Arcp::MessageTypes::JOB_RESULT) + + client.close + server_task.stop + end + end + end + + describe 'stream_result writer totals (#29)' do + it 'records totals when the agent uses the non-block writer path' do + Sync do + runtime = build_runtime( + agents: { manual_streamer: lambda { |ctx| + writer = ctx.stream_result(encoding: 'utf8') + writer.write('alpha ', more: true) + writer.write('beta', more: false) + writer.close + ctx.finish + } } + ) + client, server_task = open_pair(runtime) + handle = client.submit_job(agent: 'manual_streamer') + events = handle.subscribe(client: client).to_a + chunks = events.select { |e| e.kind == Arcp::Job::EventKind::RESULT_CHUNK } + + expect(chunks.map { |e| e.body.decoded }.join).to eq('alpha beta') + + result = handle.get_result(client: client) + expect(result.chunked?).to be(true) + expect(result.result_size).to eq('alpha beta'.bytesize) + expect(result.result_id).not_to be_nil + + client.close + server_task.stop + end + end + end + + describe 'budget enforcement (#30)' do + it 'rejects spending in a currency absent from the budget' do + Sync do + lease_manager_ref = nil + runtime = build_runtime( + agents: { ccy_mismatch: lambda { |ctx| + lease_manager_ref.try_spend!(ctx.job_id, 'EUR', BigDecimal('0.10')) + ctx.finish(result: 'never') + } } + ) + lease_manager_ref = runtime.lease_manager + + client, server_task = open_pair(runtime) + handle = client.submit_job( + agent: 'ccy_mismatch', + lease_request: Arcp::Lease::LeaseRequest.new( + capabilities: ['cost.spend'], + budget: Arcp::Lease::CostBudget.parse(['USD:1.00']) + ) + ) + handle.subscribe(client: client).to_a + expect { handle.get_result(client: client) }.to raise_error(Arcp::Errors::BudgetExhausted, /not in budget/) + + client.close + server_task.stop + end + end + + it 'remains unrestricted when no budget is attached' do + Sync do + lease_manager_ref = nil + runtime = build_runtime( + agents: { free_spender: lambda { |ctx| + lease_manager_ref.try_spend!(ctx.job_id, 'USD', BigDecimal('999.00')) + ctx.finish(result: 'unbounded') + } } + ) + lease_manager_ref = runtime.lease_manager + client, server_task = open_pair(runtime) + handle = client.submit_job(agent: 'free_spender') + handle.subscribe(client: client).to_a + expect(handle.get_result(client: client).result).to eq('unbounded') + + client.close + server_task.stop + end + end + end + + describe 'lease_constraints max_budget enforcement (#31)' do + it 'rejects a lease budget that exceeds max_budget' do + Sync do + runtime = build_runtime( + agents: { echo: ->(ctx) { ctx.finish(result: nil) } } + ) + client, server_task = open_pair(runtime) + + expect do + client.submit_job( + agent: 'echo', + lease_request: Arcp::Lease::LeaseRequest.new( + capabilities: ['cost.spend'], + budget: Arcp::Lease::CostBudget.parse(['USD:5.00']) + ), + lease_constraints: Arcp::Lease::LeaseConstraints.new( + max_budget: ['USD:1.00'] + ) + ) + end.to raise_error(Arcp::Errors::LeaseSubsetViolation) + + client.close + server_task.stop + end + end + + it 'accepts a lease budget within max_budget' do + Sync do + runtime = build_runtime( + agents: { echo: ->(ctx) { ctx.finish(result: nil) } } + ) + client, server_task = open_pair(runtime) + + handle = client.submit_job( + agent: 'echo', + lease_request: Arcp::Lease::LeaseRequest.new( + capabilities: ['cost.spend'], + budget: Arcp::Lease::CostBudget.parse(['USD:0.50']) + ), + lease_constraints: Arcp::Lease::LeaseConstraints.new( + max_budget: ['USD:1.00'] + ) + ) + expect(handle).not_to be_nil + + client.close + server_task.stop + end + end + + it 'round-trips max_budget through wire shape' do + constraints = Arcp::Lease::LeaseConstraints.new( + expires_at: nil, max_budget: ['USD:2.50', 'EUR:1.00'] + ) + restored = Arcp::Lease::LeaseConstraints.from_h(constraints.to_h) + expect(restored.max_budget.remaining('USD').to_s('F')).to eq('2.5') + expect(restored.max_budget.remaining('EUR').to_s('F')).to eq('1.0') + end + end + + describe 'client waiter safety (#32)' do + it 'raises ProtocolViolation when get_result is woken by a closed transport' do + Sync do + runtime = build_runtime( + agents: { hang: lambda { |_ctx| + # Never finishes; we close the transport mid-flight. + sleep + } } + ) + client, server_task = open_pair(runtime) + handle = client.submit_job(agent: 'hang') + + waiter = Async do + expect { client.get_result(job_id: handle.job_id) } + .to raise_error(Arcp::Errors::ProtocolViolation) + end + + Async::Task.current.sleep(0.02) + client.close + waiter.wait + server_task.stop + end + end + end + + describe 'session.bye on close (#33)' do + it 'emits a session.bye envelope before marking the client closed' do + Sync do + runtime = build_runtime + server_t, client_t = Arcp::Transport::MemoryTransport.pair + server_task = Async { runtime.accept(server_t) } + client = Arcp::Client.open(transport: client_t, auth: { 'token' => 'demo' }, client_name: 'spec') + + client.close(reason: 'done') + + bye_envelopes = client_t.sent.select { |e| e.type == Arcp::MessageTypes::SESSION_BYE } + expect(bye_envelopes.size).to eq(1) + expect(bye_envelopes.first.payload['reason']).to eq('done') + + server_task.stop + end + end + end + + describe 'list_jobs pagination stability (#38)' do + it 'keeps cursors stable when new jobs are submitted between page reads' do + Sync do + runtime = build_runtime(agents: { echo: ->(ctx) { ctx.finish(result: nil) } }) + client, server_task = open_pair(runtime) + + Array.new(3) { client.submit_job(agent: 'echo') } + .each do |h| + h.subscribe(client: client).to_a + h.get_result(client: client) + end + + page1 = runtime.job_manager.list(principal_id: 'alice', limit: 2) + expect(page1.jobs.size).to eq(2) + expect(page1.next_cursor).not_to be_nil + + # Inject another job between page reads. + client.submit_job(agent: 'echo').subscribe(client: client).to_a + + page2 = runtime.job_manager.list(principal_id: 'alice', limit: 2, cursor: page1.next_cursor) + # The third job from the original burst plus the freshly-added one fit on page 2. + ids = (page1.jobs + page2.jobs).map { |j| j['job_id'] } + expect(ids.uniq.size).to eq(ids.size) + expect(ids.size).to be >= 3 + + client.close + server_task.stop + end + end + end +end diff --git a/spec/integration/extra_branches_spec.rb b/spec/integration/extra_branches_spec.rb new file mode 100644 index 0000000..a2049a8 --- /dev/null +++ b/spec/integration/extra_branches_spec.rb @@ -0,0 +1,200 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'bigdecimal' + +# Integration spec aimed at branches that the happy-path coverage misses: +# protocol/session error replies, list_jobs filter branches, idempotency +# behavior, cancellation auth, and a few client dispatch corners. + +RSpec.describe 'extra branch coverage', type: :integration do + describe 'session.list_jobs filter branches' do + it 'narrows by status and agent prefix and respects an explicit cursor' do + Sync do + runtime = build_runtime( + agents: { echo: ->(ctx) { ctx.finish(result: nil) } } + ) + client, server_task = open_pair(runtime) + Array.new(4) { client.submit_job(agent: 'echo') } + .each do |h| + h.subscribe(client: client).to_a + h.get_result(client: client) + end + + succeeded = client.list_jobs(status: ['succeeded']).to_a + expect(succeeded.size).to eq(4) + + agent_match = client.list_jobs(agent: 'echo').to_a + expect(agent_match.size).to eq(4) + + # Sliding through the lazy enumerator with a small cursor verifies + # both the cursor passing and the next-page break. + first_page = client.list_jobs(limit: 1).first(1) + expect(first_page.size).to eq(1) + + client.close + server_task.stop + end + end + end + + describe 'job idempotency' do + it 'returns the same job id when re-submitting with the same key' do + Sync do + runtime = build_runtime( + agents: { slow: ->(ctx) { ctx.finish(result: nil) } } + ) + client, server_task = open_pair(runtime) + + h1 = client.submit_job(agent: 'slow', idempotency_key: 'k1') + h2 = client.submit_job(agent: 'slow', idempotency_key: 'k1') + expect(h2.job_id).to eq(h1.job_id) + + client.close + server_task.stop + end + end + + it 'rejects an idempotency key reused with a different agent' do + Sync do + runtime = build_runtime( + agents: { + a: ->(ctx) { ctx.finish(result: nil) }, + b: ->(ctx) { ctx.finish(result: nil) } + } + ) + client, server_task = open_pair(runtime) + + client.submit_job(agent: 'a', idempotency_key: 'k2') + expect do + client.submit_job(agent: 'b', idempotency_key: 'k2') + end.to raise_error(Arcp::Errors::DuplicateKey) + + client.close + server_task.stop + end + end + end + + describe 'cancel authorization' do + it 'rejects cancel from a different principal' do + Sync do + runtime = build_runtime( + agents: { slow: lambda { |_ctx| + loop { Async::Task.current.sleep(0.1) } + } }, + tokens: { 'a-tok' => 'alice', 'b-tok' => 'bob' } + ) + a, a_task = open_pair(runtime, auth: { 'token' => 'a-tok' }) + b, b_task = open_pair(runtime, auth: { 'token' => 'b-tok' }) + + handle = a.submit_job(agent: 'slow') + # Cancellation from bob should not succeed and the runtime should + # reply with a job error referencing the unauthorized session. + b.cancel_job(job_id: handle.job_id, reason: 'unauthorized') + + # Give the runtime a tick to process bob's cancel. + Async::Task.current.sleep(0.01) + + # alice can still cancel successfully. + handle.cancel(client: a, reason: 'done') + expect { handle.get_result(client: a) }.to raise_error(Arcp::Errors::Cancelled) + + a.close + b.close + a_task.stop + b_task.stop + end + end + end + + describe 'handshake error replies' do + it 'raises ProtocolViolation when the runtime sends an unexpected envelope' do + Sync do + server_t, client_t = Arcp::Transport::MemoryTransport.pair + bad_env = Arcp::Envelope.build( + type: Arcp::MessageTypes::SESSION_PING, + session_id: 'ses_bad', payload: { 'nonce' => 'n' } + ) + # Force a non-welcome envelope into the client's receive queue. + task = Async do + client_t.instance_variable_get(:@incoming).enqueue(bad_env) + end + task.wait + + expect do + Arcp::Client.open(transport: client_t, auth: { 'token' => 'demo' }, client_name: 'spec') + end.to raise_error(Arcp::Errors::ProtocolViolation, /expected session\.welcome/) + + server_t.close + end + end + + it 'raises ProtocolViolation when transport closes before welcome' do + Sync do + server_t, client_t = Arcp::Transport::MemoryTransport.pair + client_t.instance_variable_get(:@incoming).enqueue(nil) + + expect do + Arcp::Client.open(transport: client_t, auth: { 'token' => 'demo' }, client_name: 'spec') + end.to raise_error(Arcp::Errors::ProtocolViolation, /closed/) + + server_t.close + end + end + end + + describe 'client require_feature!' do + it 'raises UnnegotiatedFeature when ack is not in the negotiated set' do + Sync do + runtime = build_runtime(agents: { echo: ->(ctx) { ctx.finish(result: nil) } }) + client, server_task = open_pair(runtime) + + caps = client.session.capabilities + stripped = Arcp::Session::CapabilitySet.new( + features: caps.features - [Arcp::Session::Feature::ACK], + encodings: caps.encodings, agents: caps.agents + ) + client.instance_variable_set(:@session, + Arcp::Session::Info.new( + id: client.session.id, + runtime_version: client.session.runtime_version, + capabilities: stripped, + agents: client.session.agents, + heartbeat_interval_sec: client.session.heartbeat_interval_sec, + resume_token: client.session.resume_token, + resume_window_sec: client.session.resume_window_sec + )) + + expect { client.ack(1) }.to raise_error(Arcp::Errors::UnnegotiatedFeature) + + client.close + server_task.stop + end + end + end + + describe 'lease lifecycle hooks' do + it 'fires publish_error and revokes lease when an agent raises' do + Sync do + runtime = build_runtime( + agents: { broken: ->(_ctx) { raise 'boom' } } + ) + client, server_task = open_pair(runtime) + handle = client.submit_job( + agent: 'broken', + lease_request: Arcp::Lease::LeaseRequest.new( + capabilities: ['cost.spend'], + budget: Arcp::Lease::CostBudget.parse(['USD:1.00']) + ) + ) + handle.subscribe(client: client).to_a + expect { handle.get_result(client: client) }.to raise_error(Arcp::Errors::Internal) + expect(runtime.lease_manager.get(handle.job_id)).to be_nil + + client.close + server_task.stop + end + end + end +end diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index 31826a9..8666352 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -5,6 +5,8 @@ enable_coverage :branch add_filter '/spec/' add_filter '/samples/' + add_filter '/recipes/' + minimum_coverage line: 90, branch: 80 end $LOAD_PATH.unshift(File.expand_path('../lib', __dir__)) diff --git a/spec/unit/coverage_extras_spec.rb b/spec/unit/coverage_extras_spec.rb new file mode 100644 index 0000000..16b74fe --- /dev/null +++ b/spec/unit/coverage_extras_spec.rb @@ -0,0 +1,407 @@ +# frozen_string_literal: true + +require 'spec_helper' + +# Extra unit coverage targeted at the branches that the integration suite +# does not naturally exercise: tiny serializers (event-body round trips), +# envelope validation, transport edge cases, error payload helpers, and +# trace context inheritance. + +RSpec.describe 'Event body decoders' do + it 'Metric round-trips with and without a unit' do + bare = Arcp::Job::EventBody::Metric.from_h('name' => 'lat', 'value' => 1) + expect(bare.to_h).to eq({ 'name' => 'lat', 'value' => 1 }) + with_unit = Arcp::Job::EventBody::Metric.from_h('name' => 'lat', 'value' => 1, 'unit' => 'ms') + expect(with_unit.to_h['unit']).to eq('ms') + end + + it 'Log round-trips with and without fields' do + bare = Arcp::Job::EventBody::Log.from_h('level' => 'info', 'message' => 'm') + expect(bare.to_h).to eq({ 'level' => 'info', 'message' => 'm' }) + payload = Arcp::Job::EventBody::Log.from_h( + 'level' => 'info', 'message' => 'm', 'fields' => { 'a' => 1 } + ).to_h + expect(payload['fields']).to eq({ 'a' => 1 }) + end + + it 'Delegate round-trips with and without a lease' do + bare = Arcp::Job::EventBody::Delegate.from_h('child_job_id' => 'job_c', 'agent' => 'a@1') + expect(bare.to_h.key?('lease')).to be(false) + leased = Arcp::Job::EventBody::Delegate.from_h( + 'child_job_id' => 'job_c', 'agent' => 'a@1', + 'lease' => { + 'id' => 'lse_1', 'capabilities' => ['fs.read'], + 'issued_at' => '2026-01-01T00:00:00Z' + } + ) + expect(leased.to_h['lease']['id']).to eq('lse_1') + end + + it 'ToolResult ok? reflects presence of error and serializes accordingly' do + r = Arcp::Job::EventBody::ToolResult.from_h('call_id' => 'c', 'result' => { 'x' => 1 }) + expect(r.ok?).to be(true) + expect(r.to_h.key?('error')).to be(false) + e = Arcp::Job::EventBody::ToolResult.from_h('call_id' => 'c', 'error' => 'boom') + expect(e.ok?).to be(false) + expect(e.to_h.key?('result')).to be(false) + end + + it 'TraceSpan round-trips with and without optional fields' do + bare = Arcp::Job::EventBody::TraceSpan.from_h('span_id' => 's', 'name' => 'n') + expect(bare.to_h).to eq({ 'span_id' => 's', 'name' => 'n' }) + full = Arcp::Job::EventBody::TraceSpan.from_h( + 'span_id' => 's', 'name' => 'n', 'parent_span_id' => 'p', + 'start_at' => 't1', 'end_at' => 't2', 'attributes' => { 'k' => 'v' } + ) + h = full.to_h + expect(h['parent_span_id']).to eq('p') + expect(h['attributes']).to eq({ 'k' => 'v' }) + end + + it 'Status round-trips with and without optional fields' do + bare = Arcp::Job::EventBody::Status.from_h('phase' => 'starting') + expect(bare.to_h).to eq({ 'phase' => 'starting' }) + full = Arcp::Job::EventBody::Status.from_h( + 'phase' => 'starting', 'message' => 'go', 'fields' => { 'k' => 'v' } + ) + expect(full.to_h['message']).to eq('go') + end + + it 'Progress round-trips with all optional fields' do + full = Arcp::Job::EventBody::Progress.from_h( + 'current' => 5, 'total' => 10, 'units' => 'rows', 'message' => 'half' + ) + h = full.to_h + expect(h['total']).to eq(10) + expect(h['units']).to eq('rows') + end +end + +RSpec.describe 'Job request shapes' do + it 'Cancel round-trips with and without a reason' do + expect(Arcp::Job::Cancel.from_h('job_id' => 'j').to_h).to eq({ 'job_id' => 'j' }) + expect(Arcp::Job::Cancel.from_h('job_id' => 'j', 'reason' => 'x').to_h['reason']).to eq('x') + end + + it 'Subscribe round-trips with and without from_event_seq' do + bare = Arcp::Job::Subscribe.from_h('job_id' => 'j') + expect(bare.history).to be(false) + full = Arcp::Job::Subscribe.from_h('job_id' => 'j', 'from_event_seq' => 3, 'history' => true) + expect(full.to_h['from_event_seq']).to eq(3) + end + + it 'Summary round-trips with and without lease and budget' do + bare = Arcp::Job::Summary.from_h( + 'job_id' => 'j', 'agent' => 'a@1', 'status' => 'pending', 'created_at' => 't' + ) + expect(bare.to_h).not_to include('lease_expires_at') + full = Arcp::Job::Summary.from_h( + 'job_id' => 'j', 'agent' => 'a@1', 'status' => 'pending', 'created_at' => 't', + 'lease_expires_at' => 'x', 'budget_remaining' => { 'USD' => '1.0' } + ) + expect(full.to_h['budget_remaining']).to eq({ 'USD' => '1.0' }) + end +end + +RSpec.describe Arcp::Error do + it 'to_payload includes details when present and omits otherwise' do + base = Arcp::Errors::InvalidRequest.new('bad', details: { 'k' => 'v' }) + expect(base.to_payload).to include(:details, :code, :message) + bare = Arcp::Errors::InvalidRequest.new('bare') + expect(bare.to_payload.key?(:details)).to be(false) + end + + it 'to_payload includes trace_id when supplied' do + err = Arcp::Errors::InvalidRequest.new('bad') + expect(err.to_payload(trace_id: 'tid')).to include(trace_id: 'tid') + end + + it 'Errors.for falls back to Internal for unknown codes' do + expect(Arcp::Errors.for('NOT_A_CODE')).to be_a(Arcp::Errors::Internal) + end + + it 'classifies retryable and non-retryable codes' do + expect(Arcp::Errors::RETRYABLE_BY_DEFAULT).to include('AGENT_NOT_AVAILABLE') + expect(Arcp::Errors::NON_RETRYABLE_BY_DEFAULT).to include('PERMISSION_DENIED') + end +end + +RSpec.describe Arcp::Trace do + it 'returns a default Context when nothing is set' do + ctx = described_class.current + expect(ctx.trace_id).to be_nil + end + + it 'with(...) inherits trace_id and merges attributes' do + described_class.with(trace_id: 'a' * 32, attributes: { 'k' => 'v' }) do + inner = described_class.current + expect(inner.attributes).to include('k' => 'v') + expect(inner.trace_id).to eq('a' * 32) + end + expect(described_class.current.trace_id).to be_nil + end + + it 'in_span runs the supplied block and yields a span value' do + seen = nil + described_class.in_span('op', attributes: { 'a' => 1 }) do |c| + seen = c + end + expect(seen).not_to be_nil + end +end + +RSpec.describe Arcp::Session::Welcome do + it 'omits the agent inventory when absent' do + payload = { + 'runtime_name' => 'r', 'runtime_version' => '1.0', + 'capabilities' => { 'features' => ['heartbeat'], 'encodings' => ['utf8'] }, + 'heartbeat_interval_sec' => 30, + 'resume_token' => 'tok', 'resume_window_sec' => 300 + } + w = described_class.from_h(payload) + expect(w.capabilities.agents).to be_nil + end + + it 'parses the agent inventory when present' do + payload = { + 'runtime_name' => 'r', 'runtime_version' => '1.0', + 'capabilities' => { + 'features' => ['heartbeat'], 'encodings' => ['utf8'], + 'agents' => [{ 'name' => 'a', 'versions' => ['1.0.0'], 'default' => '1.0.0' }] + }, + 'heartbeat_interval_sec' => nil, 'resume_token' => nil, 'resume_window_sec' => 300 + } + w = described_class.from_h(payload) + expect(w.capabilities.agents.entries.first.name).to eq('a') + end +end + +RSpec.describe Arcp::Transport::MemoryTransport do + it 'pairs two transports and round-trips an envelope' do + a, b = described_class.pair + env = Arcp::Envelope.build(type: 'session.ping', session_id: 's', payload: {}) + a.send(env) + received = b.receive + expect(received.type).to eq('session.ping') + end + + it 'closes are idempotent' do + a, _b = described_class.pair + a.close + expect(a.closed?).to be(true) + a.close + expect(a.closed?).to be(true) + end + + it 'receive returns nil after the queue drains post-close' do + a, b = described_class.pair + a.close + expect(b.receive).to be_nil + end + + it 'send on a closed transport raises IOError' do + a, _b = described_class.pair + a.close + expect { a.send(Arcp::Envelope.build(type: 'x', session_id: 's', payload: {})) } + .to raise_error(IOError) + end +end + +RSpec.describe Arcp::Credential do + it 'redacts the value field' do + cred = described_class.new( + id: 'c1', scheme: 'bearer', value: 'sk-secret', endpoint: 'https://x' + ) + expect(cred.to_redacted_h['value']).to eq('[REDACTED]') + end + + it 'round-trips through the wire shape including profile and constraints' do + h = { + 'id' => 'c1', 'scheme' => 'bearer', 'value' => 'sk', + 'endpoint' => 'https://x', 'profile' => 'openai', + 'constraints' => { 'k' => 'v' } + } + cred = described_class.from_h(h) + expect(cred.to_h).to eq(h) + end + + it 'omits profile and empty constraints from to_h' do + cred = described_class.new(id: 'c', scheme: 'bearer', value: 'sk', endpoint: 'https://x') + h = cred.to_h + expect(h.key?('profile')).to be(false) + expect(h.key?('constraints')).to be(false) + end +end + +RSpec.describe Arcp::Session::AgentInventory do + let(:inventory) do + described_class.from_array([ + { 'name' => 'echo', 'versions' => ['1.0.0', '1.1.0'], 'default' => '1.1.0' }, + { 'name' => 'bare', 'versions' => [], 'default' => nil } + ]) + end + + it 'finds entries and lists names and versions' do + expect(inventory.find('echo').default).to eq('1.1.0') + expect(inventory.find('missing')).to be_nil + expect(inventory.names).to eq(%w[echo bare]) + expect(inventory.default_for('echo')).to eq('1.1.0') + expect(inventory.versions_for('echo')).to include('1.0.0') + expect(inventory.versions_for('missing')).to eq([]) + end + + it 'resolves agent refs against the declared versions' do + expect(inventory.resolve('echo')).to eq('echo@1.1.0') + expect(inventory.resolve('echo@1.0.0')).to eq('echo@1.0.0') + expect(inventory.resolve('echo@9.9.9')).to be_nil + expect(inventory.resolve('bare')).to be_nil + expect(inventory.resolve('missing@1')).to be_nil + end +end + +RSpec.describe Arcp::Job::AgentRef do + it 'parses a bare agent name' do + ref = described_class.parse('echo') + expect(ref.name).to eq('echo') + expect(ref.version).to be_nil + expect(ref.to_s).to eq('echo') + end + + it 'parses a versioned agent ref' do + ref = described_class.parse('echo@1.2.3') + expect(ref.to_s).to eq('echo@1.2.3') + end + + it 'rejects nil and empty names' do + expect(described_class.parse(nil)).to be_nil + expect { described_class.parse('') }.to raise_error(Arcp::Errors::InvalidRequest) + expect { described_class.parse('@1') }.to raise_error(Arcp::Errors::InvalidRequest) + end +end + +RSpec.describe Arcp::Job::Event do + it 'falls back to a frozen hash body for unknown kinds' do + e = described_class.from_h('kind' => 'mystery', 'body' => { 'a' => 1 }) + expect(e.body).to eq({ 'a' => 1 }) + expect(e.known?).to be(false) + expect(e.body).to be_frozen + end + + it 'serializes through to_h symmetrically' do + e = described_class.from_h('kind' => 'log', 'body' => { 'level' => 'info', 'message' => 'm' }) + expect(e.known?).to be(true) + expect(e.to_h).to include('kind' => 'log') + end +end + +RSpec.describe Arcp::Runtime::CredentialRegistry do + let(:store) { Arcp::Credentials::InMemoryStore.new } + let(:provisioner) { Arcp::Credentials::InMemoryProvisioner.new } + let(:registry) { described_class.new(provisioner: provisioner, store: store) } + + it 'issues credentials and records them in the store' do + cred = registry.issue_for( + job_id: 'job_a', lease: nil, agent: 'a@1', principal_id: 'alice' + ).first + expect(store.outstanding(job_id: 'job_a')).to eq([cred.id]) + end + + it 'rotate records a new credential id and revokes the old' do + issued = registry.issue_for(job_id: 'job_b', lease: nil, agent: 'a@1', principal_id: 'p').first + new_id = registry.rotate(job_id: 'job_b', credential_id: issued.id, new_value: 'sk-new') + expect(new_id).to match(/_rotated_/) + expect(store.outstanding(job_id: 'job_b')).to include(new_id) + end + + it 'revoke_all revokes every credential and forgets them on success' do + registry.issue_for(job_id: 'job_c', lease: nil, agent: 'a@1', principal_id: 'p') + revoked_count = registry.revoke_all(job_id: 'job_c') + expect(revoked_count).to be >= 1 + expect(store.outstanding(job_id: 'job_c')).to be_empty + end + + it 'reconcile_on_startup! revokes outstanding credentials from the store' do + store.record(job_id: 'job_d', credential_id: 'left_over_c1') + expect { registry.reconcile_on_startup! }.not_to raise_error + expect(store.outstanding(job_id: 'job_d')).to be_empty + end + + it 'retries a transient revoke failure once' do + flaky = Class.new do + attr_reader :calls + + def initialize = (@calls = 0) + def issue(**_) = [] + + def revoke(credential_id:) + @calls += 1 + raise 'transient' if @calls == 1 + + nil + end + end.new + r = described_class.new(provisioner: flaky, store: store) + store.record(job_id: 'job_e', credential_id: 'c-e') + count = r.revoke_all(job_id: 'job_e') + expect(flaky.calls).to eq(2) + expect(count).to eq(1) + end +end + +RSpec.describe 'Trace.current=' do + it 'sets a custom trace context for the current fiber' do + ctx = Arcp::Trace::Context.new(trace_id: 'a' * 32, span_id: 'b' * 16, attributes: { 'k' => 'v' }.freeze) + Arcp::Trace.current = ctx + expect(Arcp::Trace.current.trace_id).to eq('a' * 32) + ensure + Arcp::Trace.current = nil + end +end + +RSpec.describe Arcp::Job::EventBody::Progress do + it 'omits all optional fields when not supplied' do + p = described_class.from_h('current' => 1) + expect(p.to_h).to eq({ 'current' => 1 }) + end +end + +RSpec.describe Arcp::Job::EventBody::ResultChunk do + it 'serializes a base64 chunk through to_h' do + body = described_class.new( + result_id: 'r', chunk_seq: 0, data: 'abcd', encoding: 'base64', more: true + ) + expect(body.to_h['encoding']).to eq('base64') + end +end + +RSpec.describe Arcp::Envelope do + it 'rejects payloads of the wrong shape' do + expect do + described_class.from_h('arcp' => Arcp::PROTOCOL_VERSION, 'id' => 'i', + 'type' => 't', 'session_id' => 's', 'payload' => 'no') + end.to raise_error(Arcp::Errors::InvalidRequest, /payload must be/) + end + + it 'rejects an envelope that is not a Hash' do + expect { described_class.from_h([]) }.to raise_error(Arcp::Errors::InvalidRequest, /Hash/) + end + + it 'rejects an envelope with a non-string type' do + payload = { 'arcp' => Arcp::PROTOCOL_VERSION, 'id' => 'i', 'type' => 1, + 'session_id' => 's', 'payload' => {} } + expect { described_class.from_h(payload) }.to raise_error(Arcp::Errors::InvalidRequest, /String/) + end + + it 'rejects an envelope with a non-string session_id' do + payload = { 'arcp' => Arcp::PROTOCOL_VERSION, 'id' => 'i', 'type' => 't', + 'session_id' => 1, 'payload' => {} } + expect { described_class.from_h(payload) }.to raise_error(Arcp::Errors::InvalidRequest, /session_id/) + end + + it 'known? reports whether the wire type is registered' do + env = described_class.build(type: 'session.ping', session_id: 's', payload: {}) + expect(env.known?).to be(true) + other = described_class.build(type: 'something.unknown', session_id: 's', payload: {}) + expect(other.known?).to be(false) + end +end diff --git a/spec/unit/coverage_spec.rb b/spec/unit/coverage_spec.rb new file mode 100644 index 0000000..8daff86 --- /dev/null +++ b/spec/unit/coverage_spec.rb @@ -0,0 +1,559 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'bigdecimal' + +# Focused unit coverage for branches that integration tests don't exercise: +# transport error/EOF paths, serializer backend selection, body decoder +# error branches, event log eviction, lease manager edge cases, and the +# resume registry expiry timer. + +RSpec.describe Arcp::Serializer do + before { described_class.instance_variable_set(:@backend, :stdlib) } + + it 'round-trips JSON through the default stdlib backend' do + blob = described_class.dump({ 'a' => 1 }) + expect(described_class.load(blob)).to eq({ 'a' => 1 }) + end + + it 'returns nil for empty or nil loads' do + expect(described_class.load(nil)).to be_nil + expect(described_class.load('')).to be_nil + end + + it 'raises ArgumentError for unknown backends' do + expect { described_class.backend = :gibberish }.to raise_error(ArgumentError, /unknown serializer backend/) + end + + it 'exposes the active backend' do + expect(described_class.backend).to eq(:stdlib) + end +end + +RSpec.describe Arcp::Transport::StdioTransport do + let(:input_r) { StringIO.new('') } + let(:output_w) { StringIO.new(+'') } + + it 'sends an envelope as newline-delimited JSON' do + envelope = Arcp::Envelope.build( + type: Arcp::MessageTypes::SESSION_PING, + session_id: 'ses_x', payload: { 'nonce' => 'n' } + ) + t = described_class.new(input: input_r, output: output_w) + t.send(envelope) + expect(output_w.string).to end_with("\n") + parsed = JSON.parse(output_w.string.strip) + expect(parsed['type']).to eq(Arcp::MessageTypes::SESSION_PING) + end + + it 'returns nil and marks closed at EOF' do + t = described_class.new(input: StringIO.new(''), output: output_w) + expect(t.receive).to be_nil + expect(t.closed?).to be(true) + end + + it 'refuses to send on a closed transport' do + t = described_class.new(input: input_r, output: output_w) + t.close + expect(t.closed?).to be(true) + expect { t.send(Arcp::Envelope.build(type: 'session.ping', session_id: 's', payload: {})) } + .to raise_error(IOError, /closed/) + end + + it 'is idempotent on close and tolerates already-closed io' do + output_w.close + t = described_class.new(input: input_r, output: output_w) + expect { t.close }.not_to raise_error + expect { t.close }.not_to raise_error + end + + it 'decodes a received JSON line into an envelope' do + env = Arcp::Envelope.build(type: 'session.ping', session_id: 's', payload: {}) + input = StringIO.new("#{Arcp::Serializer.dump(env.to_h)}\n") + t = described_class.new(input: input, output: output_w) + decoded = t.receive + expect(decoded.type).to eq('session.ping') + end +end + +RSpec.describe Arcp::Transport::WebSocketTransport do + let(:fake_conn) do + Class.new do + attr_reader :written + + def initialize(messages: []) + (@queue = messages.dup + @written = [] + @closed = false) + end + + def write(s) = @written << s + def flush = nil + + def read + raise EOFError if @closed + return nil if @queue.empty? + + @queue.shift + end + + def close = (@closed = true) + end.new(messages: [Arcp::Serializer.dump(Arcp::Envelope.build( + type: 'session.ping', session_id: 's', payload: {} + ).to_h)]) + end + + it 'writes the envelope as JSON and flushes' do + t = described_class.new(connection: fake_conn) + env = Arcp::Envelope.build(type: 'session.ping', session_id: 's', payload: {}) + t.send(env) + expect(fake_conn.written.size).to eq(1) + end + + it 'returns nil at EOF and marks the transport closed' do + conn = fake_conn + conn.close + t = described_class.new(connection: conn) + expect(t.receive).to be_nil + expect(t.closed?).to be(true) + end + + it 'refuses to send on a closed transport' do + t = described_class.new(connection: fake_conn) + t.close + expect { t.send(Arcp::Envelope.build(type: 'session.ping', session_id: 's', payload: {})) } + .to raise_error(IOError, /closed/) + end + + it 'is idempotent on close' do + t = described_class.new(connection: fake_conn) + t.close + expect { t.close }.not_to raise_error + end +end + +RSpec.describe 'EventBody decoder error branches' do + it 'raises InvalidRequest for unknown result_chunk encodings' do + expect do + Arcp::Job::EventBody::ResultChunk.from_h( + 'result_id' => 'r', 'chunk_seq' => 0, 'data' => 'x', + 'encoding' => 'rot13', 'more' => false + ) + end.to raise_error(Arcp::Errors::InvalidRequest, /unknown encoding/) + end + + it 'decodes a utf8 chunk' do + body = Arcp::Job::EventBody::ResultChunk.from_h( + 'result_id' => 'r', 'chunk_seq' => 0, 'data' => 'hi', + 'encoding' => 'utf8', 'more' => false + ) + expect(body.decoded).to eq('hi') + end + + it 'decodes a base64 chunk' do + body = Arcp::Job::EventBody::ResultChunk.from_h( + 'result_id' => 'r', 'chunk_seq' => 1, + 'data' => Base64.strict_encode64('abc'), + 'encoding' => 'base64', 'more' => true + ) + expect(body.decoded).to eq('abc') + end +end + +RSpec.describe Arcp::Runtime::EventLog do + let(:clock) { Arcp::FakeClock.new } + let(:log) { described_class.new(window_sec: 30, clock: clock) } + + def env(seq, job: 'job_a') + Arcp::Envelope.build( + type: Arcp::MessageTypes::JOB_EVENT, + session_id: 'ses_x', job_id: job, event_seq: seq, payload: {} + ) + end + + it 'evicts envelopes up to a given seq' do + 3.times { |i| log.append('ses_x', env(i + 1)) } + log.evict_up_to('ses_x', 2) + expect(log.replay('ses_x').map(&:event_seq)).to eq([3]) + expect(log.floor('ses_x')).to eq(2) + end + + it 'replays events past a from_event_seq cursor' do + 3.times { |i| log.append('ses_x', env(i + 1)) } + expect(log.replay('ses_x', from_event_seq: 2).map(&:event_seq)).to eq([2, 3]) + end + + it 'expires entries past the window' do + log.append('ses_x', env(1)) + clock.advance(31) + log.expire! + expect(log.buffer_size('ses_x')).to eq(0) + expect(log.job_buffer_size('job_a')).to eq(0) + end + + it 'tracks job-indexed replay independently of the producing session' do + log.append('ses_producer', env(1, job: 'job_b')) + log.append('ses_producer', env(2, job: 'job_b')) + expect(log.replay_job('job_b').map(&:event_seq)).to eq([1, 2]) + expect(log.replay('ses_observer')).to be_empty + end +end + +RSpec.describe Arcp::Runtime::ResumeRegistry do + let(:clock) { Arcp::FakeClock.new } + let(:registry) { described_class.new(window_sec: 10, clock: clock) } + + it 'tracks a token and returns the entry on lookup' do + registry.register(token: 'tok', session_id: 'ses_x', principal_id: 'p1') + expect(registry.lookup('tok').session_id).to eq('ses_x') + end + + it 'evicts disconnected entries past the window' do + registry.register(token: 'tok', session_id: 'ses_x', principal_id: 'p1') + registry.mark_disconnected('tok', last_processed_seq: 3) + clock.advance(11) + expect(registry.lookup('tok')).to be_nil + end + + it 'leaves connected entries indefinitely' do + registry.register(token: 'tok', session_id: 'ses_x', principal_id: 'p1') + clock.advance(120) + expect(registry.lookup('tok')).not_to be_nil + end + + it 'mark_reconnected clears the disconnect timer' do + registry.register(token: 'tok', session_id: 'ses_x', principal_id: 'p1') + registry.mark_disconnected('tok') + registry.mark_reconnected('tok') + clock.advance(120) + expect(registry.lookup('tok')).not_to be_nil + end + + it 'forget removes the entry' do + registry.register(token: 'tok', session_id: 'ses_x', principal_id: 'p1') + registry.forget('tok') + expect(registry.lookup('tok')).to be_nil + end + + it 'expire! drops only disconnected-expired entries' do + registry.register(token: 'a', session_id: 'sa', principal_id: 'p') + registry.register(token: 'b', session_id: 'sb', principal_id: 'p') + registry.mark_disconnected('a') + clock.advance(11) + registry.expire! + expect(registry.lookup('a')).to be_nil + expect(registry.lookup('b')).not_to be_nil + end +end + +RSpec.describe Arcp::Runtime::LeaseManager do + let(:clock) { Arcp::FakeClock.new } + let(:lm) { described_class.new(clock: clock) } + + def lease(currencies: nil, capabilities: ['cost.spend'], expires_at: nil, model_use: nil) + budget = currencies ? Arcp::Lease::CostBudget.parse(currencies) : nil + Arcp::Lease::Lease.new( + id: 'lse_x', capabilities: capabilities, budget: budget, + model_use: model_use, expires_at: expires_at, + issued_at: clock.now.iso8601 + ) + end + + it 'returns unlimited spend when no lease is registered' do + expect(lm.try_spend!('job_unknown', 'USD', BigDecimal('1.00'))).to be(true) + end + + it 'allows any currency on a lease with no budget' do + lm.register('job_a', lease) + expect(lm.try_spend!('job_a', 'XYZ', BigDecimal('99'))).to be(true) + end + + it 'denies an unlisted currency on a budgeted lease' do + lm.register('job_b', lease(currencies: ['USD:1.00'])) + expect { lm.try_spend!('job_b', 'EUR', BigDecimal('0.10')) } + .to raise_error(Arcp::Errors::BudgetExhausted, /not in budget/) + end + + it 'spends exactly the remaining balance then raises on the next call' do + lm.register('job_c', lease(currencies: ['USD:0.50'])) + expect(lm.try_spend!('job_c', 'USD', BigDecimal('0.50'))).to be(true) + expect { lm.try_spend!('job_c', 'USD', BigDecimal('0.01')) } + .to raise_error(Arcp::Errors::BudgetExhausted, /exhausted/) + end + + it 'raises LeaseExpired for an expired lease' do + expired = lease(expires_at: (clock.now - 1).iso8601) + lm.register('job_d', expired) + expect { lm.check!('job_d', capability: 'cost.spend') } + .to raise_error(Arcp::Errors::LeaseExpired) + end + + it 'raises PermissionDenied when capability is not in the lease' do + lm.register('job_e', lease(capabilities: ['fs.read'])) + expect { lm.check!('job_e', capability: 'net.fetch') } + .to raise_error(Arcp::Errors::PermissionDenied) + end + + it 'is a no-op for jobs without leases' do + expect { lm.check!('job_none', capability: 'fs.read') }.not_to raise_error + end + + it 'allows model.use matches and denies misses' do + lm.register('job_f', lease(model_use: ['gpt-4*'])) + expect(lm.check_model!('job_f', model_id: 'gpt-4-turbo')).to be(true) + expect { lm.check_model!('job_f', model_id: 'claude-opus') } + .to raise_error(Arcp::Errors::PermissionDenied) + end + + it 'exposes the remaining snapshot' do + lm.register('job_g', lease(currencies: ['USD:2.00'])) + lm.try_spend!('job_g', 'USD', BigDecimal('0.75')) + expect(lm.remaining('job_g')['USD'].to_s('F')).to eq('1.25') + expect(lm.remaining('job_missing')).to eq({}) + end +end + +RSpec.describe Arcp::Runtime::SubscriptionManager do + let(:sm) { described_class.new } + + it 'rejects attach by a non-owning principal' do + sm.register_owner('job_a', 'alice', 'ses_a', Object.new) + expect { sm.attach('job_a', 'bob', 'ses_b', Object.new) } + .to raise_error(Arcp::Errors::PermissionDenied) + end + + it 'detaches a subscriber and stops receiving fanout' do + a_q = [] + b_q = [] + sm.register_owner('job_a', 'alice', 'ses_a', + Object.new.tap { |o| o.define_singleton_method(:enqueue) { |e| a_q << e } }) + sm.attach('job_a', 'alice', 'ses_b', + Object.new.tap { |o| o.define_singleton_method(:enqueue) { |e| b_q << e } }) + + env = Arcp::Envelope.build(type: 'job.event', session_id: 's', job_id: 'job_a', event_seq: 1, payload: {}) + sm.fanout('job_a', env) + expect(a_q.size).to eq(1) + expect(b_q.size).to eq(1) + + sm.detach('job_a', 'ses_b') + sm.fanout('job_a', env) + expect(b_q.size).to eq(1) + end + + it 'reports the owner principal' do + sm.register_owner('job_a', 'alice', 'ses_a', Object.new) + expect(sm.owner_of('job_a')).to eq('alice') + end + + it 'rebinds a session id to a new outbox across jobs' do + old_q = [] + new_q = [] + old_outbox = Object.new.tap { |o| o.define_singleton_method(:enqueue) { |e| old_q << e } } + new_outbox = Object.new.tap { |o| o.define_singleton_method(:enqueue) { |e| new_q << e } } + + sm.register_owner('job_a', 'alice', 'ses_a', old_outbox) + sm.rebind_session('ses_a', new_outbox) + + env = Arcp::Envelope.build(type: 'job.event', session_id: 's', job_id: 'job_a', event_seq: 1, payload: {}) + sm.fanout('job_a', env) + expect(new_q.size).to eq(1) + expect(old_q).to be_empty + end +end + +RSpec.describe Arcp::Auth::Bearer do + let(:principal) { Arcp::Auth::Principal.new(id: 'alice', name: 'alice', scopes: [].freeze) } + + it 'returns nil for missing tokens' do + expect(described_class.new(tokens: {}).verify(nil)).to be_nil + expect(described_class.new(tokens: {}).verify('unknown')).to be_nil + end + + it 'accepts a Principal value' do + verifier = described_class.new(tokens: { 'tok' => principal }) + expect(verifier.verify('tok').id).to eq('alice') + end + + it 'accepts a String shorthand' do + verifier = described_class.new(tokens: { 'tok' => 'alice' }) + expect(verifier.verify('tok').id).to eq('alice') + end + + it 'accepts a Hash shape' do + verifier = described_class.new(tokens: { 'tok' => { 'id' => 'alice', 'name' => 'A', 'scopes' => ['x'] } }) + p = verifier.verify('tok') + expect(p.id).to eq('alice') + expect(p.scopes).to eq(['x']) + end + + it 'from_token builds a single-token verifier' do + verifier = described_class.from_token('tok', principal_id: 'a') + expect(verifier.verify('tok').id).to eq('a') + end +end + +RSpec.describe Arcp::Credentials do + it 'translates a budget-exhausted upstream error' do + upstream = Class.new(StandardError) do + define_method(:code) { 'budget_exhausted' } + end.new('upstream said no') + translated = described_class.translate_upstream_error(upstream) + expect(translated).to be_a(Arcp::Errors::BudgetExhausted) + end + + it 'translates a 402-status upstream error' do + upstream = Class.new(StandardError) do + define_method(:status) { 402 } + end.new('payment') + translated = described_class.translate_upstream_error(upstream) + expect(translated).to be_a(Arcp::Errors::BudgetExhausted) + end + + it 'passes other errors through unchanged' do + err = StandardError.new('other') + expect(described_class.translate_upstream_error(err)).to be(err) + end + + it 'InMemoryProvisioner issues and revokes credentials' do + p = described_class::InMemoryProvisioner.new + cred = p.issue(lease: nil, job_id: 'job_a', agent: 'echo@1', principal_id: 'alice').first + expect(cred.id).to eq('cred_job_a_0') + p.revoke(credential_id: cred.id) + expect(p.revoked).to eq([cred.id]) + end + + it 'InMemoryStore records, lists, and forgets credential ids' do + s = described_class::InMemoryStore.new + s.record(job_id: 'job_a', credential_id: 'c1') + s.record(job_id: 'job_a', credential_id: 'c1') # dedupe + s.record(job_id: 'job_a', credential_id: 'c2') + expect(s.outstanding(job_id: 'job_a').sort).to eq(%w[c1 c2]) + s.forget(job_id: 'job_a', credential_id: 'c1') + expect(s.outstanding(job_id: 'job_a')).to eq(['c2']) + s.forget(job_id: 'job_a', credential_id: 'c2') + expect(s.all_outstanding).to be_empty + end +end + +RSpec.describe Arcp::Lease::LeaseConstraints do + it 'rejects malformed max_budget values' do + expect { described_class.new(max_budget: 42) } + .to raise_error(Arcp::Errors::InvalidRequest, /max_budget/) + end + + it 'accepts CostBudget directly' do + cb = Arcp::Lease::CostBudget.parse(['USD:1.00']) + c = described_class.new(max_budget: cb) + expect(c.max_budget).to eq(cb) + end + + it 'accepts an array of "CCY:amount" entries' do + c = described_class.new(max_budget: ['USD:1.00']) + expect(c.max_budget.remaining('USD').to_s('F')).to eq('1.0') + end + + it 'enforce_max_budget! is a no-op when max_budget is nil' do + expect { described_class.new.enforce_max_budget!(Arcp::Lease::CostBudget.parse(['USD:50.00'])) } + .not_to raise_error + end + + it 'enforce_max_budget! is a no-op when requested is nil' do + expect { described_class.new(max_budget: ['USD:1.00']).enforce_max_budget!(nil) }.not_to raise_error + end + + it 'rejects currencies not declared in max_budget' do + c = described_class.new(max_budget: ['USD:1.00']) + expect { c.enforce_max_budget!(Arcp::Lease::CostBudget.parse(['EUR:0.10'])) } + .to raise_error(Arcp::Errors::LeaseSubsetViolation) + end + + it 'validates UTC expires_at' do + expect { described_class.new(expires_at: '2026-01-01T00:00:00+02:00').validate! } + .to raise_error(Arcp::Errors::InvalidRequest, /must be UTC/) + end +end + +RSpec.describe 'JobContext#stream_result writer totals (#29)' do + let(:sink) do + Class.new do + attr_reader :events, :result, :error + + def initialize + (@events = [] + @result = nil + @error = nil) + end + + def runtime = nil + + def publish_event(_jid, event) + @events << event + @events.size + end + + def publish_result(_jid, result) = (@result = result) + def publish_error(_jid, err) = (@error = err) + end.new + end + + it 'records totals when the writer is closed before finish (non-block path)' do + ctx = Arcp::Runtime::JobContext.new(job_id: 'job_a', agent: 'a@1', input: nil, lease: nil, sink: sink) + writer = ctx.stream_result(encoding: 'utf8') + writer.write('hello ', more: true) + writer.write('world', more: false) + writer.close + ctx.finish + + expect(sink.result.chunked?).to be(true) + expect(sink.result.result_size).to eq('hello world'.bytesize) + expect(sink.events.size).to eq(2) + end + + it 'is idempotent on double close' do + ctx = Arcp::Runtime::JobContext.new(job_id: 'job_b', agent: 'a@1', input: nil, lease: nil, sink: sink) + writer = ctx.stream_result + writer.write('x', more: false) + writer.close + expect { writer.close }.not_to raise_error + end + + it 'rejects inline result mixed with chunk stream' do + ctx = Arcp::Runtime::JobContext.new(job_id: 'job_c', agent: 'a@1', input: nil, lease: nil, sink: sink) + writer = ctx.stream_result + writer.write('x', more: false) + writer.close + expect { ctx.finish(result: 'inline') } + .to raise_error(Arcp::Errors::ProtocolViolation, /cannot mix/) + end +end + +RSpec.describe Arcp::Runtime::JobManager do + let(:runtime) do + Arcp::Runtime::Runtime.new( + auth_verifier: Arcp::Auth::Bearer.from_token('tok', principal_id: 'alice'), + heartbeat_interval_sec: nil + ) + end + let(:jm) { runtime.job_manager } + + it 'raises AgentNotAvailable for an unregistered agent' do + expect { jm.resolve_agent('missing') }.to raise_error(Arcp::Errors::AgentNotAvailable) + end + + it 'raises AgentVersionNotAvailable for a missing pinned version' do + jm.register_agent(name: 'a', versions: ['1.0.0'], default: '1.0.0', handler: ->(_) {}) + expect { jm.resolve_agent('a@2.0.0') }.to raise_error(Arcp::Errors::AgentVersionNotAvailable) + end + + it 'list with an empty cursor walks from the oldest job and respects limit' do + response = jm.list(principal_id: 'alice', limit: 10) + expect(response.next_cursor).to be_nil + expect(response.jobs).to eq([]) + end + + it 'decode_cursor falls back to 0 for malformed strings' do + # Public-ish behavior probe via send to keep the private helpers covered. + expect(jm.send(:decode_cursor, 'not-an-int')).to eq(0) + expect(jm.send(:decode_cursor, '')).to eq(0) + end +end