Class: Terret::Tools::Approvals

Inherits:
Hames::Service
  • Object
show all
Defined in:
lib/terret/approvals.rb

Overview

ctx — durable human-in-the-loop gating (plan §6.3, §12 M6). An execute-stage middleware parks calls whose definition demands a decision, appends durable approval/requested, and resumes when a matching approval/resolved lands in the log (the socket's approve/deny frames append exactly that). Both sides are durable, so a parked call survives a process death: on resume the gate finds the recorded verdict and never parks. It gates on tools/execute rather than tools/pre_execute so pre_execute vetoes (the per-agent AllowList) settle a call before a human is ever asked.

Instance Method Summary collapse

Instance Method Details

#deny_pending!(session_id, reason: "cancelled") ⇒ Object

Cancel's escape hatch: deny everything parked for a session, durably. Each denial is an ordinary approval/resolved append, so an in-process parked fiber unparks through the same listener as a socket verdict, and a restart-orphaned request settles for good.



67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/terret/approvals.rb', line 67

def deny_pending!(session_id, reason: "cancelled")
  pending(session_id).each do |call_id|
    @ctx[:sessions].append(session_id, "approval/resolved",
                           { call_id: call_id, verdict: "denied", reason: reason })
  end
  # The durable denials above wake one waiter per unique pending id
  # through the resolved listener. A provider that reused a call id within
  # one parallel batch parked more than one fiber on that id, though, and
  # a cancel means all of them: sweep whatever is still parked for this
  # session onto the denial they share, so no fiber is left holding the
  # barrier open. Waiters the appends already woke are gone from the map,
  # so this touches only the ones a single durable denial could not reach.
  drain_session_waiters(session_id, { verdict: "denied", reason: reason })
end

#pending(session_id) ⇒ Object

Log-derived: requested without a matching resolved, within the OPEN turn. The in-memory waiter map is never consulted — after a restart it is empty while the log still knows what is owed. Nothing is pending once the turn that asked has closed: a request its turn outlived was settled by that turn ending, and a provider is free to reuse the call id afterwards.



50
51
52
53
54
55
56
57
58
59
 
# File 'lib/terret/approvals.rb', line 50

def pending(session_id)
  events = open_turn(session_id)
  resolved = events.filter_map { |e| e.payload[:call_id] if e.type == "approval/resolved" }
  events.filter_map do |e|
    next unless e.type == "approval/requested"
    next if resolved.include?(e.payload[:call_id])

    e.payload[:call_id]
  end
end

#pending?(session_id, call_id) ⇒ Boolean

Returns:

  • (Boolean) 


61
 
# File 'lib/terret/approvals.rb', line 61

def pending?(session_id, call_id) = pending(session_id).include?(call_id)

#start(ctx) ⇒ Object 



19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/terret/approvals.rb', line 19

def start(ctx)
  @ctx = ctx
  # A unique per-park token => { session_id:, call_id:, queue: }. Keyed by
  # the token rather than by [session_id, call_id] because provider
  # tool-call ids are NOT contractually unique: two calls of one parallel
  # batch can arrive sharing an id, and keying by it let the second park
  # overwrite the first's queue so a single verdict woke only the survivor
  # and the first fiber parked forever, hanging the barrier. Each park now
  # owns its own entry and its own wake. The DURABLE correlation
  # (approval/requested and /resolved, matched in the log on call_id +
  # name + args) is unchanged; only this in-memory map needed unique keys.
  @waiting = {}
  @waiting_mutex = Mutex.new # tests resolve from another thread; wake_one scans
  @park_seq = 0

  ctx.on("tools/execute") do |call, next_|
    gate(call, next_)
  end
  ctx.on("session/event") do |ev|
    next unless ev.type == "approval/resolved"

    wake_one(ev.session_id, ev.payload[:call_id], ev.payload)
  end
end