diff --git a/docs/sessions.md b/docs/sessions.md index 7d7ddb75..34fea133 100644 --- a/docs/sessions.md +++ b/docs/sessions.md @@ -261,10 +261,11 @@ normal spawn path and `continue_conversation` moves on to the next candidate. > `HOME` the subprocess will see — `options.env`'s `HOME` when it sets one): > > - `.credentials.json`, with the OAuth `refreshToken` removed so the resumed -> subprocess can't consume it. On macOS with the default config dir and no +> subprocess can't consume it. On macOS with no > `ANTHROPIC_API_KEY`/`CLAUDE_CODE_OAUTH_TOKEN`, the credentials come from -> the Keychain entry when one exists (the redirected config dir would -> otherwise miss it). +> the CLI's Keychain entry for your config dir when one exists (the +> redirected config dir would otherwise miss it) — with a custom +> `CLAUDE_CONFIG_DIR`, only when that directory has no `.credentials.json`. > - `.claude.json` (from `$CLAUDE_CONFIG_DIR/.claude.json` when set, else > `~/.claude.json`). > - User `settings.json` and `cowork_settings.json` — so `apiKeyHelper`, `env`, @@ -276,7 +277,10 @@ normal spawn path and `continue_conversation` moves on to the next candidate. > Everything else in your config dir is **not** visible to the subprocess — > notably user `CLAUDE.md`, `agents/`, `skills/`, and `plugins/` (so, with the > plugin keys stripped, user plugins are off) — so a store-backed resume can -> still behave differently from a plain `resume:` of the same session. +> still behave differently from a plain `resume:` of the same session. The +> project's auto-memory is part of that: the temp config dir has no +> `projects//memory/`, so a store-backed resume neither loads the memory +> you already have nor keeps what the session writes to it. > Project-level `.claude/*` still applies (it resolves from `cwd`), and > hooks/options passed programmatically via `ClaudeAgentOptions` are unaffected. > Seeded files are written owner-only (`0600`); a missing source file is simply @@ -286,8 +290,16 @@ normal spawn path and `continue_conversation` moves on to the next candidate. > (terminal append failures — timeouts immediately, other failures after up to > three attempts — surfaced as `MirrorErrorMessage`): > the store copy is then incomplete and the temp dir holds the only copy of the -> dropped turns, so the SDK scrubs the credential copies, keeps the transcripts, -> and warns with the preserved path so you can import them into the store. +> dropped turns, so the SDK moves it into a fresh private directory next to it +> (`claude-preserved-resume-*`), keeps the transcripts (`projects/`) there, +> deletes everything else — the seeded credential and settings copies and +> whatever the CLI wrote beside them, such as its `backups/` copy of +> `.claude.json` — and warns with the new path so you can import them into the +> store. If what it moved is not the directory the SDK created (the path had +> been replaced, for instance by a symlink), or it cannot be moved, the SDK +> deletes nothing and the warning says the scrub was skipped and why; if an +> entry cannot be removed, the warning says the scrub failed and names what is +> left. ### Implementing an adapter diff --git a/examples/session_stores/README.md b/examples/session_stores/README.md index c27cba48..95edcedb 100644 --- a/examples/session_stores/README.md +++ b/examples/session_stores/README.md @@ -131,7 +131,11 @@ through the relevant items below. - Pass a connection backed by a pool sized ≥ expected concurrent sessions for heavy use; don't share one connection with request-handler code that holds it. -- `jsonb` reorders keys — contract-safe, but don't byte-compare entries. +- Keep `entry` a `json` column. `jsonb` rejects the `\u0000` escape, so one + entry with a NUL character (binary tool output) fails the whole `INSERT` and + the mirror drops that batch. A table created by an earlier copy of this + adapter still has `jsonb`, and `create_schema` does not alter an existing + table: run the migration under [`json`, not `jsonb`](#json-not-jsonb). - Add a retention job (`DELETE WHERE mtime < ...`) — the table grows unbounded. --- @@ -250,7 +254,7 @@ CREATE TABLE IF NOT EXISTS claude_session_store ( session_id text NOT NULL, subpath text NOT NULL DEFAULT '', seq bigserial, - entry jsonb NOT NULL, + entry json NOT NULL, mtime bigint NOT NULL, PRIMARY KEY (project_key, session_id, subpath, seq) ); @@ -262,14 +266,30 @@ CREATE INDEX IF NOT EXISTS claude_session_store_list_idx `#load` is `SELECT entry ... ORDER BY seq`. The empty string is the `subpath` sentinel for the main transcript so the composite primary key is total. -### JSONB key ordering +### `json`, not `jsonb` -Entries are stored as `jsonb`, which **reorders object keys** on read-back. This -is explicitly allowed by the SessionStore contract — `#load` requires -*deep-equal*, not *byte-equal*, returns. The Ruby SDK reads entry fields by key -from the parsed Hash (it never does a byte/prefix scan over stored entries), so -reordering is transparent. Use a `json` or `text` column if you need byte-stable -storage. +Entries are stored in a `json` column. `jsonb` rejects the `\u0000` escape, and +transcripts do carry it: a NUL character in binary tool output is serialized +that way. One such entry fails the whole multi-row `INSERT`, so the mirror +drops every entry of that batch (surfaced as a `MirrorErrorMessage`) and a +later store-backed resume replays a transcript with that turn missing. `json` +keeps the text exactly as given, key order included. The adapter only ever +reads whole entries, so it needs none of `jsonb`'s operators or indexes. If you +add SQL that looks inside entries, note that `->` and `->>` still fail on an +entry that contains the escape. + +`create_schema` is `CREATE TABLE IF NOT EXISTS`, so it does **not** change a +table created by an earlier copy of this adapter. That table keeps its `jsonb` +column, and keeps rejecting those entries, until you migrate it: + +```sql +ALTER TABLE claude_session_store ALTER COLUMN entry TYPE json USING entry::json; +``` + +The statement rewrites the table under an `ACCESS EXCLUSIVE` lock, so pick a +quiet moment for a large one. Rows written while the column was `jsonb` come +back with the key order `jsonb` gave them, which the SessionStore contract +allows: `#load` requires *deep-equal*, not *byte-equal*, returns. `#delete` cascades to subpath rows; it is only invoked via `ClaudeAgentSDK.delete_session(session_id:, session_store: store)`. diff --git a/examples/session_stores/postgres_session_store.rb b/examples/session_stores/postgres_session_store.rb index 08a10a76..668e617c 100644 --- a/examples/session_stores/postgres_session_store.rb +++ b/examples/session_stores/postgres_session_store.rb @@ -34,7 +34,7 @@ # session_id text NOT NULL, # subpath text NOT NULL DEFAULT '', # seq bigserial, -# entry jsonb NOT NULL, +# entry json NOT NULL, # mtime bigint NOT NULL, # PRIMARY KEY (project_key, session_id, subpath, seq) # ); @@ -53,11 +53,18 @@ # many concurrent sessions, use one adapter instance (with its own # connection) per session instead. # -# JSONB key ordering: entries are stored as `jsonb`, which REORDERS object keys -# on read-back. This is explicitly allowed by the SessionStore contract — #load -# requires deep-equal, not byte-equal, returns. The Ruby SDK reads entry fields -# by key from the parsed Hash (never a byte/prefix scan), so reordering is -# transparent. Use a `json` or `text` column if you need byte-stable storage. +# Column type: entries are stored as `json`, not `jsonb`. `jsonb` rejects the +# \u0000 escape, and transcripts do carry it (a NUL character in binary tool +# output): one such entry fails the whole multi-row INSERT, so the mirror drops +# every entry of that batch. `json` keeps the text as given, key order +# included; this adapter only ever reads whole entries, so it needs none of +# jsonb's operators or indexes. +# +# A table created by an earlier copy of this adapter keeps its `jsonb` column: +# #create_schema is CREATE TABLE IF NOT EXISTS and never alters an existing +# table. Migrate it explicitly: +# +# ALTER TABLE claude_session_store ALTER COLUMN entry TYPE json USING entry::json; # # Retention: this adapter never deletes rows on its own. Add a scheduled # DELETE ... WHERE mtime < $cutoff (or partition by mtime). @@ -103,7 +110,7 @@ def create_schema session_id text NOT NULL, subpath text NOT NULL DEFAULT '', seq bigserial, - entry jsonb NOT NULL, + entry json NOT NULL, mtime bigint NOT NULL, PRIMARY KEY (project_key, session_id, subpath, seq) ); @@ -128,7 +135,7 @@ def append(key, entries) db_exec_params( <<~SQL, INSERT INTO #{@table} (project_key, session_id, subpath, entry, mtime) - SELECT $1, $2, $3, e::jsonb, $5 + SELECT $1, $2, $3, e::json, $5 FROM unnest($4::text[]) WITH ORDINALITY AS t(e, ord) ORDER BY ord SQL @@ -145,8 +152,8 @@ def load(key) ) return nil if result.ntuples.zero? - # The pg gem returns jsonb as its JSON text; parse each row. (If a type map - # decodes jsonb to a Hash already, pass it through.) + # The pg gem returns json as its JSON text; parse each row. (If a type map + # decodes json to a Hash already, pass it through.) out = result.map do |row| v = row['entry'] v.is_a?(String) ? JSON.parse(v) : v diff --git a/lib/claude_agent_sdk/session_resume.rb b/lib/claude_agent_sdk/session_resume.rb index 8b6d04c1..92ce8a5c 100644 --- a/lib/claude_agent_sdk/session_resume.rb +++ b/lib/claude_agent_sdk/session_resume.rb @@ -1,6 +1,7 @@ # frozen_string_literal: true require 'json' +require 'digest' require 'fileutils' require 'tmpdir' require 'open3' @@ -22,31 +23,167 @@ module ClaudeAgentSDK class MaterializedResume attr_reader :config_dir, :resume_session_id - def initialize(config_dir:, resume_session_id:) + MIRROR_DROPPED = 'Claude SDK: transcript mirror dropped batches; the session store copy is incomplete.' + SKIPPED = 'Scrubbing was skipped and nothing was deleted:' + # Prefix of the private directory a preserved temp dir is moved into. Not + # "claude-resume-": that one is for directories the SDK may still delete. + STAGING_PREFIX = 'claude-preserved-resume-' + private_constant :MIRROR_DROPPED, :SKIPPED, :STAGING_PREFIX + + # +root_identity+ is SessionResume.directory_identity of +config_dir+ as + # the SDK created it; materialize_resume_session takes it right after + # mkdtemp. Without it, it is taken here. + def initialize(config_dir:, resume_session_id:, root_identity: nil) @config_dir = config_dir @resume_session_id = resume_session_id + @root_identity = root_identity || SessionResume.directory_identity(config_dir) + @kept = false end # Best-effort removal of the temp config dir (never raises). + # + # Does nothing once preserve_transcripts was called, however that went. A + # teardown can run twice — a disconnect after one that was cut short — + # and the second time nothing remembers that the mirror dropped batches: + # it asks for a cleanup of the directory holding the only copy of them. + # Leaving copies that were not scrubbed yet is the lesser harm. def cleanup + return if @kept + SessionResume.rmtree_with_retry(@config_dir) end # Teardown when the transcript mirror dropped batches: the CLI's # authoritative transcript lives in this temp dir, and the store copy is # missing the dropped turns — deleting the dir would permanently lose - # them. Keep the transcripts (projects/), remove the redacted credential - # copies, and tell the user where the data is so they can import it into - # the store manually. Never raises. + # them. Keep the transcripts (projects/), delete every other entry, and + # tell the user where the data is so they can import it into the store + # manually. Never raises. + # + # An allow-list on purpose. Besides the files the SDK seeds + # (.credentials.json, .claude.json, settings.json, cowork_settings.json) + # the CLI writes its own: at startup it saves the seeded .claude.json — + # which can hold MCP header secrets — as backups/.claude.json.backup.. + # A list of names to delete missed that one and would miss the next. + # + # Deleting "everything else" must never reach outside the directory the + # SDK created, and the CLI's tools can write into that directory — + # sandboxed ones too — up to and during this call: they can turn the + # directory, or anything in it, into a symlink between two steps taken + # here. So a path that was checked is never resolved again: + # + # 1. The directory is moved, in one rename, into a fresh private directory + # next to it. A rename moves a symlink itself, never its target, and it + # takes the directory away from the path its writers know. (A read-only + # directory cannot be moved; it is made writable first, through a + # handle that has to be the directory the SDK created.) + # 2. What was moved is then examined: it has to be a real directory with + # the device and inode recorded when the SDK created it. If it is not, + # or it could not be moved, nothing is deleted and the warning says why. + # 3. Its entries are deleted by SessionResume::Scrubber, which takes each + # one out of the tree before it looks at it. + # + # config_dir is the new location afterwards, the one the warning names. def preserve_transcripts - ['.credentials.json', '.claude.json', 'settings.json', 'cowork_settings.json'].each do |name| - FileUtils.rm_f(File.join(@config_dir, name)) - end - warn 'Claude SDK: transcript mirror dropped batches; the session store copy is incomplete. ' \ - "Preserving the session transcript under #{File.join(@config_dir, 'projects')} instead of " \ - 'deleting it — import it into your session store, then remove the directory.' + @kept = true # first: whatever happens below, #cleanup must not delete it + original = @config_dir + announced = false + warn "#{MIRROR_DROPPED} #{move_aside_and_scrub(original)}" + announced = true rescue StandardError => e warn "Claude SDK: failed to scrub preserved transcript dir #{@config_dir}: #{e.message}" + announced = true + ensure + # Cut short by something that is not a StandardError (a cancellation, a + # signal). #cleanup leaves the directory alone from now on, so say what + # is left to do with it. + warn "#{MIRROR_DROPPED} #{interrupted_notice(original)}" unless announced + end + + private + + # What is left to say when the scrub was cut short: where the transcript + # is, that copies which were not removed yet may be left, and which + # directory to remove once the transcript is imported. That is the private + # directory once the move happened (config_dir changed with it; the trash + # is in there too) and the temp dir itself before — never its parent, + # which would be the system's temp directory. + def interrupted_notice(original) + holding = @config_dir == original ? @config_dir : File.dirname(@config_dir) + "Scrubbing was interrupted; the session transcript is under #{File.join(@config_dir, 'projects')}. Copies of " \ + "your credentials and settings that were not removed yet may be left under #{holding} — import the " \ + "transcript into your session store, then remove #{holding}." + end + + # Returns the rest of the preservation warning. +original+ is config_dir + # as it is when preserve_transcripts starts. + def move_aside_and_scrub(original) + staging, failure = move_aside(original) + if failure + return "#{SKIPPED} #{original} could not be moved aside (#{failure.message}). If it is still there, the " \ + "session transcript is under #{File.join(original, 'projects')}, next to copies of your " \ + 'credentials and settings that were not removed.' + end + + @config_dir = File.join(staging, File.basename(original)) + unless @root_identity && SessionResume.directory_identity(@config_dir) == @root_identity + return "#{SKIPPED} what was at #{original} is not the directory the SDK created (it had been replaced). " \ + "It is now at #{@config_dir}; check what it holds before importing anything from it." + end + + preserved = "Preserving the session transcript under #{File.join(@config_dir, 'projects')} instead of " \ + "deleting it — import it into your session store, then remove #{staging}." + [preserved, scrub_all_but_projects(staging)].compact.join(' ') + end + + # Step 1. Returns [private directory the root now lives in, nil], or + # [nil, error] when it stayed where it was. + def move_aside(original) + staging = nil + make_root_movable(original) + staging = Dir.mktmpdir(STAGING_PREFIX, File.dirname(original)) + File.rename(original, File.join(staging, File.basename(original))) + [staging, nil] + rescue SystemCallError, SessionResume::Scrubber::Changed => e + Dir.rmdir(staging) if staging + [nil, e] + end + + # Moving a directory to another parent takes write permission on the + # directory itself. Skipping the scrub for a read-only one would leave the + # credential copies behind, so it is made accessible first — through a + # handle that has to be the directory the SDK created. + def make_root_movable(root) + stat = File.lstat(root) + return if !stat.directory? || SessionResume::Scrubber.owner_rwx?(stat) + + SessionResume::Scrubber.make_accessible(root, @root_identity) + end + + # Step 3. Deletes every entry of the moved root but projects/, then checks + # that nothing else is left. Returns nil, or what the warning has to add. + def scrub_all_but_projects(staging) + trash = Dir.mktmpdir('scrub-', staging) + scrubber = SessionResume::Scrubber.new(trash) + errors = {} + Dir.children(@config_dir).each do |name| + next if name == 'projects' + + scrubber.remove(@config_dir, name) + rescue SessionResume::Scrubber::Changed => e + errors[name] = e + break # something is rearranging the directory under the scrub + rescue SystemCallError => e + errors[name] = e + end + Dir.rmdir(trash) if errors.empty? + + left = (Dir.children(@config_dir) - ['projects'] + errors.keys).uniq.sort + return if left.empty? + + reason = errors.values_at(*left).compact.first&.message + "Scrubbing failed: could not remove #{left.join(', ')}#{" (#{reason})" if reason}. What is left is under " \ + "#{staging} and can hold copies of your credentials and settings." end end @@ -75,6 +212,10 @@ module SessionResume # rubocop:disable Metrics/ModuleLength -- resume materializ KEYCHAIN_SERVICE_NAME = 'Claude Code-credentials' KEYCHAIN_TIMEOUT_SECONDS = 5 + # The default ClaudeAgentOptions gives load_timeout_ms; used when a caller + # set the attribute back to nil. + DEFAULT_LOAD_TIMEOUT_MS = 60_000 + # SystemCallError classes that indicate a transiently-held handle (Windows # AV/indexer scanning a freshly-written file) or a recoverable resource # shortage (file-table exhaustion) rather than a permanent failure. EMFILE/ @@ -89,9 +230,11 @@ module SessionResume # rubocop:disable Metrics/ModuleLength -- resume materializ # Return a copy of +options+ repointed at a materialized temp config dir: # CLAUDE_CONFIG_DIR in env, resume set to the materialized session id, and # continue_conversation cleared (already resolved to a concrete session id). + # options.env reads nil once a caller set it back to nil (the constructor + # default is {}); that means no overrides. def apply_materialized_options(options, materialized) options.dup_with( - env: options.env.merge('CLAUDE_CONFIG_DIR' => materialized.config_dir.to_s), + env: (options.env || {}).merge('CLAUDE_CONFIG_DIR' => materialized.config_dir.to_s), resume: materialized.resume_session_id, continue_conversation: false ) @@ -123,10 +266,11 @@ def build_mirror_batcher(store:, env:, on_error:, eager: false, callback_wrapper # exception or the timeout) if a store call fails or times out. def materialize_resume_session(options) # rubocop:disable Metrics/AbcSize -- materialization sequence kept in order store = options.session_store - return nil if store.nil? - return nil if options.resume.nil? && !options.continue_conversation + return nil if store.nil? || (options.resume.nil? && !options.continue_conversation) - timeout_s = options.load_timeout_ms / 1000.0 + # load_timeout_ms reads nil once a caller set it back to nil; the + # constructor default applies then. 0 is a valid (immediate) timeout. + timeout_s = (options.load_timeout_ms || DEFAULT_LOAD_TIMEOUT_MS) / 1000.0 # Probed ONCE at materialization entry (the resume path's construction # point) so an invalid callback_scheduling declaration fails fast here, # before any store IO or temp-dir work. @@ -152,6 +296,9 @@ def materialize_resume_session(options) # rubocop:disable Metrics/AbcSize -- mat session_id, lines = resolved tmp_base = Dir.mktmpdir('claude-resume-') begin + # Taken now: a teardown that deletes inside the directory checks that + # the path still leads to this one (MaterializedResume#preserve_transcripts). + root_identity = directory_identity(tmp_base) project_dir = File.join(tmp_base, 'projects', project_key) FileUtils.mkdir_p(project_dir) write_jsonl(File.join(project_dir, "#{session_id}.jsonl"), lines) @@ -172,7 +319,7 @@ def materialize_resume_session(options) # rubocop:disable Metrics/AbcSize -- mat raise end - MaterializedResume.new(config_dir: tmp_base, resume_session_id: session_id) + MaterializedResume.new(config_dir: tmp_base, resume_session_id: session_id, root_identity: root_identity) end # -- Helpers -- @@ -357,16 +504,23 @@ def copy_auth_files(tmp_base, opt_env) # rubocop:disable Metrics/AbcSize, Metric # read_if_present returns raw bytes; the credentials path parses and # re-serializes JSON, so hand it a UTF-8-tagged string (invalid bytes # simply fail to parse and get written through, as before). - creds_bytes = source_config_dir && read_if_present(File.join(source_config_dir, '.credentials.json')) + creds_path = source_config_dir && File.join(source_config_dir, '.credentials.json') + creds_bytes = creds_path && read_if_present(creds_path) creds_json = creds_bytes&.dup&.force_encoding(Encoding::UTF_8) - # macOS default keeps OAuth tokens in the Keychain, not a file. Redirecting - # CLAUDE_CONFIG_DIR changes the Keychain service suffix so the subprocess's - # lookup misses; populate the plaintext file from the parent's Keychain. - # Skipped when env-based auth or a custom config dir is already in play. - if caller_config_dir.nil? && env_value(opt_env, 'ANTHROPIC_API_KEY').nil? && - env_value(opt_env, 'CLAUDE_CODE_OAUTH_TOKEN').nil? - keychain = read_keychain_credentials + # macOS keeps OAuth tokens in the Keychain, not in a file, under a + # service name that depends on the config dir (see + # keychain_service_name). Redirecting CLAUDE_CONFIG_DIR changes that + # name, so the subprocess's own lookup misses; populate the plaintext + # file from the caller's entry instead. Skipped under env-based auth. + # + # Default config dir: a Keychain hit overrides the file. Custom config + # dir: the Keychain is consulted only when the directory has no + # .credentials.json at all — which of the two the CLI prefers when both + # exist is not established, so a file that is there keeps winning. + if env_value(opt_env, 'ANTHROPIC_API_KEY').nil? && env_value(opt_env, 'CLAUDE_CODE_OAUTH_TOKEN').nil? && + (caller_config_dir.nil? || !File.exist?(creds_path)) + keychain = read_keychain_credentials(caller_config_dir) creds_json = keychain unless keychain.nil? end @@ -536,9 +690,11 @@ def redacted_credentials(creds_json) creds_json end - # Read OAuth credentials JSON from the macOS Keychain (default service name). - # Best-effort — returns nil on any error or non-macOS platforms. - def read_keychain_credentials + # Read OAuth credentials JSON from the macOS Keychain entry the CLI keeps + # for +config_dir+ (the caller's CLAUDE_CONFIG_DIR; nil for the default + # config dir). Best-effort — returns nil on any error or non-macOS + # platforms. + def read_keychain_credentials(config_dir) return nil unless RbConfig::CONFIG['host_os'].match?(/darwin/) user = (ENV['USER'] && !ENV['USER'].empty? ? ENV['USER'] : nil) || begin @@ -549,7 +705,7 @@ def read_keychain_credentials end stdout, status = capture_with_timeout( - ['security', 'find-generic-password', '-a', user, '-w', '-s', KEYCHAIN_SERVICE_NAME], + ['security', 'find-generic-password', '-a', user, '-w', '-s', keychain_service_name(config_dir)], KEYCHAIN_TIMEOUT_SECONDS ) return nil if status.nil? || !status.success? @@ -560,6 +716,21 @@ def read_keychain_credentials nil end + # The Keychain service the CLI stores its credentials under. With the + # default config dir that is KEYCHAIN_SERVICE_NAME; with a custom + # CLAUDE_CONFIG_DIR the CLI appends the first 8 hex digits of the SHA-256 + # of that directory — the string the environment carries (no realpath), + # NFC-normalized. The bytes are tagged UTF-8 first, which is how the CLI + # reads them: under a C locale Ruby tags a non-ASCII ENV value as binary, + # and unicode_normalize rejects that. + def keychain_service_name(config_dir) + return KEYCHAIN_SERVICE_NAME if config_dir.nil? + + dir = config_dir.to_s.dup.force_encoding(Encoding::UTF_8) + dir = dir.unicode_normalize(:nfc) if dir.valid_encoding? + "#{KEYCHAIN_SERVICE_NAME}-#{Digest::SHA256.hexdigest(dir)[0, 8]}" + end + # Run a command with a hard timeout, draining stdout on a side thread and # SIGKILL-ing on deadline (Timeout.timeout is unsafe under the fiber # scheduler). Returns [stdout, status] or [nil, nil] on timeout/error. @@ -686,6 +857,143 @@ def resolve_dir(dir) File.expand_path(dir) end + # [device, inode] of +path+ itself when it is a real directory — lstat, so + # a symlink is not followed and does not count — else nil. Taken when the + # temp config dir is created; MaterializedResume compares it before it + # deletes anything inside. + def directory_identity(path) + stat = File.lstat(path) + stat.directory? ? [stat.dev, stat.ino] : nil + rescue SystemCallError, TypeError + nil + end + + # Deletes entries of a directory tree that something else may still be + # writing into, without ever following a symlink — not even one that takes + # an entry's place while this runs. + # + # Looking at an entry and then acting on it by path leaves a gap: what + # lstat called a directory can be a symlink by the time it is listed, + # chmodded or descended into, and a path through it then leads outside the + # tree. So an entry is taken out first — renamed into +trash+, a private + # directory nothing else knows; a rename moves a symlink itself, never its + # target. Only then is the entry examined, and unlinked or, for a + # directory, emptied the same way one level at a time. Every path used is + # therefore an entry of the directory handed to #remove (which the caller + # vouches for), trash/ or trash//. + # + # A directory can refuse: moving one to another parent takes write + # permission on the directory itself, emptying it takes read, write and + # search. It is then made accessible (see .make_accessible); nothing else + # is ever chmodded. + class Scrubber + # An entry was no longer what an earlier look at it had found. + class Changed < StandardError; end + + OWNER_RWX = 0o700 + + # How many times a directory that keeps receiving entries is emptied + # before rmdir's "not empty" is left to report it. + EMPTYING_PASSES = 3 + + def self.owner_rwx?(stat) + stat.mode.allbits?(OWNER_RWX) + end + + # chmod 0700 the directory at +path+ without following a symlink: the + # mode is set through a handle (fchmod), and only once that handle + # proved to be a directory this user owns whose [device, inode] is + # +identity+. The path is looked at again afterwards. Raises Changed + # when either look finds something else, and SystemCallError when the + # directory cannot be opened (its owner cannot read it) or chmodded. + def self.make_accessible(path, identity) + File.open(path, File::RDONLY | File::NOFOLLOW | File::NONBLOCK) do |handle| + stat = handle.stat + unless stat.directory? && [stat.dev, stat.ino] == identity + raise Changed, "#{path} is not the directory it was a moment ago" + end + raise Errno::EPERM, path unless stat.owned? + + handle.chmod(OWNER_RWX) + end + return if SessionResume.directory_identity(path) == identity + + raise Changed, "#{path} was replaced while it was being made writable" + rescue Errno::ELOOP, Errno::EMLINK + # What O_NOFOLLOW answers for a symlink (EMLINK on some BSDs). + raise Changed, "#{path} became a symlink" + end + + def initialize(trash) + @trash = trash + @taken = 0 + end + + # Remove the entry +name+ of +dir+ and everything under it. Raises + # SystemCallError when something cannot be removed, and Changed when an + # entry was swapped while it was being made accessible. + # + # Nothing here recurses: this runs on the reactor, and a tree can be + # deeper than a fiber's stack. Emptying a directory moves its entries + # into the trash, which is flat, so what is still to be removed is a + # list of trash paths, whatever depth they came from. + def remove(dir, name) + pending = [take(dir, name)] + until pending.empty? + path = pending.pop + stat = File.lstat(path) + next File.unlink(path) unless stat.directory? + + self.class.make_accessible(path, [stat.dev, stat.ino]) unless self.class.owner_rwx?(stat) + pending.concat(empty_and_remove(path)) + end + end + + private + + # Empty the directory at +path+ into the trash and remove it; returns + # where its entries are now. + # + # Something that still holds the directory open can write into it + # meanwhile, and a listing cannot tell: the entry may arrive right after + # it, and the listing may well have been empty. rmdir is what notices + # (ENOTEMPTY; EEXIST on some systems). The directory then gets another + # pass, EMPTYING_PASSES in all; the last refusal is the caller's to + # report. + def empty_and_remove(path) + taken = [] + passes = 0 + begin + passes += 1 + Dir.children(path).each { |child| taken << take(path, child) } + Dir.rmdir(path) + rescue Errno::ENOTEMPTY, Errno::EEXIST + retry if passes < EMPTYING_PASSES + raise + end + taken + end + + # Move dir/name into the trash, whatever it is, and return its new path. + def take(dir, name) + source = File.join(dir, name) + target = File.join(@trash, (@taken += 1).to_s) + begin + File.rename(source, target) + rescue Errno::EACCES, Errno::EPERM + # A directory that is not writable cannot be moved to another + # parent. It is looked at immediately before it is repaired; + # whatever else refuses (the parent, a file) is left as it is. + stat = File.lstat(source) + raise if !stat.directory? || self.class.owner_rwx?(stat) + + self.class.make_accessible(source, [stat.dev, stat.ino]) + File.rename(source, target) + end + target + end + end + # Best-effort recursive removal with retries on transient lock errors # (Windows AV/indexer). Never raises. The temp dir holds an access token, so # the final sweep matters for not leaking secrets. @@ -782,7 +1090,7 @@ def env_value(opt_env, name) private_class_method :load_candidate, :resolve_continue_candidate, :with_timeout, :write_jsonl, :copy_auth_files, :write_redacted_credentials, :read_keychain_credentials, - :capture_with_timeout, :materialize_subkeys, :write_subagent_files, + :keychain_service_name, :capture_with_timeout, :materialize_subkeys, :write_subagent_files, :resolve_dir, :read_if_present, :chmod_owner_only, :copy_if_present, :env_value, :strip_settings_for_resume, :parse_settings_bytes, :mask_surrogate_escapes, :redacted_credentials, :encode_candidate, :encode_jsonl_lines, :encode_entry, diff --git a/lib/claude_agent_sdk/transcript_mirror_batcher.rb b/lib/claude_agent_sdk/transcript_mirror_batcher.rb index d2f4cb58..682e1908 100644 --- a/lib/claude_agent_sdk/transcript_mirror_batcher.rb +++ b/lib/claude_agent_sdk/transcript_mirror_batcher.rb @@ -31,8 +31,8 @@ module ClaudeAgentSDK # (which surfaces them as a MirrorErrorMessage). Adapters should dedupe by # entry["uuid"] when present, since a retried batch may overlap a prior write. # - # The semaphore serializes appends (FIFO, and every drain detaches its batch - # immediately before queueing on it, so append order matches enqueue order), + # The semaphore serializes appends (FIFO, and every drain detaches the whole + # buffer once it holds the lock, so append order matches enqueue order), # but a #send that exceeds send_timeout is abandoned (its worker thread keeps # running) and the next drain proceeds, so two #append calls for the SAME # key can briefly overlap. SessionStore#append must be thread-safe per key @@ -107,14 +107,16 @@ def initialize(store:, projects_dir:, on_error:, send_timeout: SEND_TIMEOUT_SECO # # Frames still buffered once #close has started count too. Query#close # stops the read task (the background drainer's parent) right after - # #close returns, so a frame the read loop enqueued during the close - # window — or any below-threshold frame after it — is never appended. - # The old per-frame drain detached it into a parked task whose - # cancellation counted it; with coalescing it stays in @pending, so it is - # accounted here. Evaluated lazily: frames the drainer still manages to - # deliver before the check are not drops, and nothing past #close has to - # run for a stranded frame to be counted. (A plain ivar/Array#empty? read - # is safe cross-thread under the GVL, like @dropped_batches.) + # #close returns, so a frame the read loop enqueued after #close detached + # — during its final append, or any below-threshold frame after it — is + # never appended. (One that arrives while #close is still waiting for the + # lock goes out with #close's own batch.) The old per-frame drain detached + # such a frame into a parked task whose cancellation counted it; with + # coalescing it stays in @pending, so it is accounted here. Evaluated + # lazily: frames the drainer still manages to deliver before the check + # are not drops, and nothing past #close has to run for a stranded frame + # to be counted. (A plain ivar/Array#empty? read is safe cross-thread + # under the GVL, like @dropped_batches.) def batches_dropped? @dropped_batches.positive? || (@closed && !@pending.empty?) end @@ -150,9 +152,10 @@ def flush drain end - # Final flush before teardown. Never raises. Bounded: it waits behind at - # most the in-flight append, and frames arriving after it detached are - # not chased (they count as dropped unless delivered; #batches_dropped?). + # Final flush before teardown. Never raises. Bounded: it waits its turn on + # the lock, takes whatever is pending at that moment, and does not chase + # frames that arrive after that (they count as dropped unless delivered; + # #batches_dropped?). def close @closed = true flush @@ -168,16 +171,16 @@ def over_threshold? # Fire-and-forget on the reactor. The drainer loops until the buffer is # back under the thresholds, so while it is live every later frame is - # simply buffered and coalesced into its next #drain — live background - # tasks <= 1 and detached batches <= 1 + one per #flush / #close caller - # parked on @lock, independent of frame rate and store latency. + # simply buffered and coalesced into the next #drain — live background + # tasks <= 1 and detached batches <= 1 (only the holder of @lock has + # one), independent of frame rate and store latency. # # No lost wakeup: the loop's final over_threshold? check and the ensure # clearing the flag run with no suspension point between them, so an # #enqueue that saw the flag set is always observed by that check. - # Every drainer (this one, #flush, #close) keeps detach-then-acquire, so - # detach order == @lock FIFO order and append ordering holds. #drain - # never raises. + # Every drainer (this one, #flush, #close) detaches inside @lock and + # takes everything pending at that moment, so batches leave in @lock FIFO + # order and append ordering holds. #drain never raises. def schedule_background_drain(task) return if @background_drain_live @@ -194,25 +197,29 @@ def schedule_background_drain(task) raise end - # Detach the pending buffer, await any prior flush, then send. Detaching - # before acquiring the lock lets #enqueue keep accumulating into a fresh - # buffer while a prior flush is in flight. Never raises. + # Await any prior flush, detach the pending buffer, then send. The buffer + # is detached INSIDE the lock: a drain cancelled while it waits for the + # lock has taken nothing, so its frames stay in @pending for the next + # drain. (It used to detach first, and a #flush cancelled while queued + # behind an in-flight append lost its batch — upstream Python PR #1289.) + # #enqueue never takes the lock, so it keeps accumulating while a flush + # is in flight. Never raises. def drain - items = @pending - @pending = [] - @pending_entries = 0 - @pending_bytes = 0 - errors = [] - # Cancellation (Async::Stop — not a StandardError) delivered while - # waiting on the lock or inside the append's thread join loses the - # detached items without either rescue firing. Teardown would then read - # batches_dropped? as false and delete a materialized resume dir holding - # the only copy of these entries — count the batch as dropped unless the - # flush path ran to completion (do_flush counts its own failures). - accounted = items.empty? + # Cancellation (Async::Stop — not a StandardError) delivered inside the + # append's thread join loses the detached items without either rescue + # firing. Teardown would then read batches_dropped? as false and delete + # a materialized resume dir holding the only copy of these entries — + # count the batch as dropped unless the flush path ran to completion + # (do_flush counts its own failures). Until a batch is detached there + # is nothing to account for. + accounted = true begin @lock.acquire do + items = @pending + @pending = [] + @pending_entries = 0 + @pending_bytes = 0 # Emptiness is checked INSIDE the lock (matching the Python batcher): # an empty #flush/#close still serializes behind any in-flight or # queued drain, so they are true barriers — at result-yield and at @@ -220,6 +227,7 @@ def drain # the read task while a detached batch is still being appended. next if items.empty? + accounted = false begin do_flush(items, errors) rescue StandardError => e @@ -349,10 +357,10 @@ def invoke_append(key, entries) [:error, e] end - # Sleep that yields the reactor when one is active, else a plain sleep. + # Kernel#sleep is scheduler-aware: it parks only this fiber when a reactor + # is active and blocks the thread otherwise. def sleep_backoff(seconds) - task = Async::Task.current? - task ? task.sleep(seconds) : sleep(seconds) + sleep(seconds) end def deep_stringify(obj) diff --git a/spec/examples/postgres_session_store_payload_spec.rb b/spec/examples/postgres_session_store_payload_spec.rb new file mode 100644 index 00000000..e95e5ccf --- /dev/null +++ b/spec/examples/postgres_session_store_payload_spec.rb @@ -0,0 +1,84 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'securerandom' + +# Gated live spec, like postgres_session_store_spec.rb: runs only when the `pg` +# gem is installed (optional :examples Bundler group) AND +# SESSION_STORE_POSTGRES_URL points at a reachable server. Each example uses a +# random-suffixed table and DROPs it on teardown. +postgres_url = ENV.fetch('SESSION_STORE_POSTGRES_URL', nil) +postgres_reachable = begin + require 'pg' + require_relative '../../examples/session_stores/postgres_session_store' + if postgres_url + probe = PG.connect(postgres_url) + probe.exec('SELECT 1') + probe.close + true + else + false + end +rescue LoadError, StandardError + false +end + +# Transcripts carry NUL characters (binary tool output). JSON.generate writes +# one as the six-character escape for U+0000, which PostgreSQL's `jsonb` input +# rejects and `json` stores as given. One such entry used to fail the whole +# multi-row INSERT, so the mirror dropped every entry of that batch. +RSpec.describe 'PostgresSessionStore with a NUL character in a transcript entry', if: postgres_reachable do + let(:conn) { PG.connect(postgres_url) } + let(:table) { "cst_test_#{SecureRandom.hex(6)}" } + let(:store) { PostgresSessionStore.new(conn: conn, table: table) } + let(:key) { { 'project_key' => '-Users-dev-app', 'session_id' => SecureRandom.uuid } } + let(:binary_output) do + { 'type' => 'user', 'uuid' => 'u-nul', + 'message' => { 'role' => 'user', + 'content' => [{ 'type' => 'tool_result', 'tool_use_id' => 'toolu_01', + 'content' => "ELF#{0.chr}#{1.chr}binary" }] } } + end + + after do + conn.exec("DROP TABLE IF EXISTS #{table}") + conn.close + end + + def turn(uuid) + { 'type' => 'user', 'uuid' => uuid, 'message' => { 'role' => 'user', 'content' => "turn #{uuid}" } } + end + + it 'stores the entry, and the clean entries appended with it' do + store.create_schema + batch = [turn('u-1'), binary_output, turn('u-2')] + + store.append(key, batch) + + expect(store.load(key)).to eq(batch) + end + + # #create_schema is CREATE TABLE IF NOT EXISTS: a table created by an earlier + # copy of the adapter keeps its jsonb column until it is migrated. This is + # the migration the README gives. + it 'still rejects it on a table with the old jsonb column, until that column is migrated' do + conn.exec(<<~SQL) + CREATE TABLE #{table} ( + project_key text NOT NULL, + session_id text NOT NULL, + subpath text NOT NULL DEFAULT '', + seq bigserial, + entry jsonb NOT NULL, + mtime bigint NOT NULL, + PRIMARY KEY (project_key, session_id, subpath, seq) + ) + SQL + store.create_schema # leaves the existing table as it is + store.append(key, [turn('u-1')]) + expect { store.append(key, [binary_output]) }.to raise_error(PG::UntranslatableCharacter) + + conn.exec("ALTER TABLE #{table} ALTER COLUMN entry TYPE json USING entry::json") + + store.append(key, [binary_output]) + expect(store.load(key)).to eq([turn('u-1'), binary_output]) + end +end diff --git a/spec/integration/store_backed_resume_integration_spec.rb b/spec/integration/store_backed_resume_integration_spec.rb new file mode 100644 index 00000000..eb013bf9 --- /dev/null +++ b/spec/integration/store_backed_resume_integration_spec.rb @@ -0,0 +1,81 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'securerandom' +require 'tmpdir' +require 'json' +require 'fileutils' + +RSpec.describe 'Store-backed resume through the real Claude CLI', :integration do + # Gated by RUN_INTEGRATION and self-skipping, like real_cli_integration_spec.rb: + # this spawns the real `claude` CLI and makes two small, budget-capped API calls. + before do + skip 'Claude CLI is not available on PATH' unless system('command -v claude >/dev/null 2>&1') + skip 'ANTHROPIC_API_KEY is required for real CLI integration tests' if ENV['ANTHROPIC_API_KEY'].to_s.empty? + end + + # A disposable config dir, so the first turn's local transcript lands where + # the example can delete it before resuming. No auth is copied: the CLI + # authenticates with ANTHROPIC_API_KEY, which also keeps the SDK's macOS + # Keychain bridge out of the picture. + around do |example| + previous_config_dir = ENV.fetch('CLAUDE_CONFIG_DIR', nil) + Dir.mktmpdir('cas-store-resume') do |directory| + @config_dir = File.join(directory, 'config') + @project_dir = File.join(directory, 'project') + FileUtils.mkdir_p(@project_dir) + ENV['CLAUDE_CONFIG_DIR'] = @config_dir + example.run + end + ensure + if previous_config_dir + ENV['CLAUDE_CONFIG_DIR'] = previous_config_dir + else + ENV.delete('CLAUDE_CONFIG_DIR') + end + end + + def run_turn(prompt, options) + messages = [] + ClaudeAgentSDK.query(prompt: prompt, options: options) { |message| messages << message } + + result = messages.grep(ClaudeAgentSDK::ResultMessage).last + expect(result).to be_a(ClaudeAgentSDK::ResultMessage) + expect(result.is_error).to be(false) + expect(messages.grep(ClaudeAgentSDK::MirrorErrorMessage)).to be_empty + result + end + + it 'mirrors a turn into a SessionStore, then resumes the session from the store alone' do + store = ClaudeAgentSDK::InMemorySessionStore.new + marker = "MARK-#{SecureRandom.hex(4)}" + options = ClaudeAgentSDK::ClaudeAgentOptions.new(model: 'haiku', setting_sources: [], tools: [], + max_budget_usd: 0.05, cwd: @project_dir, session_store: store) + materialized = [] + allow(ClaudeAgentSDK::SessionResume).to receive(:materialize_resume_session).and_wrap_original do |original, opts| + original.call(opts).tap { |resume| materialized << resume if resume } + end + + first = run_turn("Remember this code for later: #{marker}. Reply with exactly: OK", options) + key = { 'project_key' => ClaudeAgentSDK.project_key_for_directory(@project_dir), 'session_id' => first.session_id } + mirrored = store.load(key) + expect(JSON.generate(mirrored)).to include(marker) # the turn reached the store + expect(materialized).to be_empty # nothing to resume yet + + # Leave the store as the only copy: what the CLI resumes below is what the + # SDK materialized from it, not the transcript the first turn left on disk. + FileUtils.rm_rf(File.join(@config_dir, 'projects')) + + second = run_turn('What code did I ask you to remember? Reply with the code only.', + options.dup_with(resume: first.session_id)) + + expect(second.session_id).to eq(first.session_id) + expect(second.result.to_s).to include(marker) # the history came back + expect(materialized.length).to eq(1) + expect(File).not_to exist(materialized.first.config_dir) # temp config dir (credential copies) removed + resumed = store.load(key) + expect(resumed.length).to be > mirrored.length # the resumed turn was mirrored on top ... + uuids = resumed.filter_map { |entry| entry['uuid'] } + expect(uuids.uniq.length).to eq(uuids.length) # ... and the materialized history was not mirrored again + end +end diff --git a/spec/support/keychain_bridge_stub.rb b/spec/support/keychain_bridge_stub.rb new file mode 100644 index 00000000..093f6aa6 --- /dev/null +++ b/spec/support/keychain_bridge_stub.rb @@ -0,0 +1,21 @@ +# frozen_string_literal: true + +# Store-backed resume bridges the macOS Keychain: unless the caller +# authenticates through the environment, materialization runs +# `security find-generic-password` for the config dir's entry. An example that +# reaches SessionResume.materialize_resume_session without stubbing that — +# any example pairing a session_store with resume — would read the developer's +# real credentials on macOS, copy them into a temp dir and depend on the state +# of the local Keychain. +# +# So the bridge is off for every example by default. The examples that test +# the bridge itself opt in with `keychain: true`, on the example or its group, +# and replace the layer below it (capture_with_timeout) instead: see +# spec/unit/session_resume_keychain_spec.rb, which also pins this default. +RSpec.configure do |config| + config.before do |example| + next if example.metadata[:keychain] + + allow(ClaudeAgentSDK::SessionResume).to receive(:read_keychain_credentials).and_return(nil) + end +end diff --git a/spec/unit/materialized_resume_spec.rb b/spec/unit/materialized_resume_spec.rb new file mode 100644 index 00000000..7f2f6c99 --- /dev/null +++ b/spec/unit/materialized_resume_spec.rb @@ -0,0 +1,783 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'json' +require 'tmpdir' +require 'fileutils' +require 'stringio' + +RSpec.describe ClaudeAgentSDK::MaterializedResume do + # When the mirror dropped batches the materialized config dir is kept, since + # it holds the only copy of the dropped turns. It must then hold nothing but + # those transcripts, and getting there must not delete or change anything + # outside it, whatever is done to the directory in the meantime: the CLI's + # tools can write into it, sandboxed ones included. + describe '#preserve_transcripts' do + # One scratch directory per example holds the temp config dir, the private + # directory the SDK moves it into (created next to it) and the directories + # nothing may touch. Nothing lands in the real $TMPDIR. + let(:scratch) { Dir.mktmpdir('preserve-spec-') } + let(:config_dir) { File.join(scratch, 'claude-resume-under-test') } + let(:outside) { File.join(scratch, 'outside') } + let(:sid) { '11111111-1111-4111-8111-111111111111' } + let(:materialized) { described_class.new(config_dir: config_dir, resume_session_id: sid) } + + # What the SDK seeds from the caller's .claude.json: it can carry MCP + # header secrets. + let(:claude_json) do + JSON.generate('oauthAccount' => { 'emailAddress' => 'user@example.invalid' }, + 'mcpServers' => { 'internal' => { 'type' => 'http', 'url' => 'https://mcp.example.invalid', + 'headers' => { 'Authorization' => 'Bearer MCP-HEADER-SECRET' } } }) + end + + # The four files the SDK seeds, then what a real CLI (2.1.286) left beside + # them after a store-backed resume — at startup it saves the seeded + # .claude.json as backups/.claude.json.backup. before rewriting + # it — and a file name nobody has heard of yet. + let(:secrets) do + { + '.credentials.json' => '{"claudeAiOauth":{"accessToken":"ACCESS-TOKEN"}}', + '.claude.json' => claude_json, + 'settings.json' => '{"env":{"INTERNAL_API_KEY":"SETTINGS-SECRET"}}', + 'cowork_settings.json' => '{"env":{"INTERNAL_API_KEY":"SETTINGS-SECRET"}}', + 'backups/.claude.json.backup.1790881328533' => claude_json, + 'cache/model-catalog/x-cc.json' => '{}', + 'written-by-a-later-cli.json' => claude_json + } + end + let(:transcripts) do + { + "projects/-proj/#{sid}.jsonl" => %({"type":"user","uuid":"u1","sessionId":"#{sid}"}\n), + "projects/-proj/#{sid}/subagents/agent-a1.jsonl" => %({"type":"user","uuid":"s1","isSidechain":true}\n), + "projects/-proj/#{sid}/subagents/agent-a1.meta.json" => '{"agentType":"worker"}' + } + end + + before do + Dir.mkdir(config_dir, 0o700) + write_tree(config_dir, secrets.merge(transcripts)) + Dir.mkdir(File.join(config_dir, 'sessions')) # the CLI leaves this one empty + write_tree(outside, 'not-ours.txt' => 'keep me') + File.symlink(outside, File.join(config_dir, 'linked-elsewhere')) + materialized # built now, like the SDK does: it records which directory this is + end + + # Stubs are still in place while this runs. A wrapper that waits for a + # call its example never got to must not make its swap now, in the middle + # of the teardown: the scratch directory would stay behind in $TMPDIR. + after do + { File => %i[lstat open rename unlink], Dir => %i[children rmdir] }.each do |receiver, wrapped| + wrapped.each { |name| allow(receiver).to receive(name).and_call_original } + end + unlock(scratch) + FileUtils.remove_entry(scratch) + end + + def write_tree(root, files) + files.each do |relative, content| + path = File.join(root, relative) + FileUtils.mkdir_p(File.dirname(path)) + File.write(path, content) + end + end + + # Give every real directory under +path+ its owner bits back (never + # through a symlink), so the scratch tree can be removed whatever an + # example left read-only. + def unlock(path) + return unless File.lstat(path).directory? + + File.chmod(0o700, path) + Dir.children(path).each { |name| unlock(File.join(path, name)) } + end + + def stderr_of + original = $stderr + $stderr = StringIO.new + yield + $stderr.string + ensure + $stderr = original + end + + # relative path => content, for every regular file under +dir+. + def tree(dir) + files = Dir.glob('**/*', File::FNM_DOTMATCH, base: dir).select { |relative| File.file?(File.join(dir, relative)) } + files.sort.to_h { |relative| [relative, File.read(File.join(dir, relative))] } + end + + def mode_of(path) + format('%o', File.stat(path).mode & 0o777) + end + + # Runs preserve_transcripts and returns its warning. + def preserve + stderr_of { materialized.preserve_transcripts } + end + + # Runs preserve_transcripts while +method+ of +receiver+ raises something + # that is not a StandardError — what a cancellation (Async::Stop) or a + # signal looks like — and returns the warning. The wrapper is disarmed + # afterwards: the examples and the teardown use the same methods. + def preserve_interrupted_at(receiver, method) + cancellation = Class.new(Exception) # rubocop:disable Lint/InheritException -- see above + interrupting = true + allow(receiver).to receive(method).and_wrap_original do |original, *args| + raise cancellation if interrupting + + original.call(*args) + end + stderr_of { expect { materialized.preserve_transcripts }.to raise_error(cancellation) } + ensure + interrupting = false + end + + # Where the directory lives once it was moved, and the private directory + # it was moved into. + def preserved + materialized.config_dir + end + + def staging + File.dirname(preserved) + end + + let(:skipped) { /transcript mirror dropped batches.*Scrubbing was skipped and nothing was deleted/ } + + # Dir.children answers in the filesystem's order, and that order differs + # from one system to the next: an example that passed on APFS failed on + # ext4. Every example that wraps Dir.children runs with the names in each + # of these orders, so that none owes its result to the machine it runs on. + listing_orders = { + 'sorted' => :sort.to_proc, + 'in reverse order' => ->(names) { names.sort.reverse }, + 'shuffled' => ->(names) { names.sort.shuffle(random: Random.new(RSpec.configuration.seed)) } + }.freeze + + # Makes Dir.children answer with the names in +order+. A block, if given, is + # called with the path and those names before they are returned. + def list_children(order) + allow(Dir).to receive(:children).and_wrap_original do |original, path| + order.call(original.call(path)).tap { |names| yield path, names if block_given? } + end + end + + def in_trash?(path) + File.basename(File.dirname(path)).start_with?('scrub-') + end + + it 'moves the directory into a private one next to it, keeps projects/ and deletes every other entry' do + warning = preserve + + expect(File.symlink?(config_dir) || File.exist?(config_dir)).to be(false) + expect(File.basename(staging)).to start_with('claude-preserved-resume-') + expect(Dir.children(scratch)).to contain_exactly(File.basename(staging), 'outside') + expect(mode_of(staging)).to eq('700') + expect(Dir.children(staging)).to eq(['claude-resume-under-test']) + expect(Dir.children(preserved)).to eq(['projects']) + expect(tree(preserved)).to eq(transcripts) + expect(warning).to match(/transcript mirror dropped batches.*Preserving the session transcript under /) + expect(warning).to match(/under #{Regexp.escape(File.join(preserved, 'projects'))} .*then remove #{Regexp.escape(staging)}\./) + expect(warning).not_to match(/Scrubbing/) + end + + it 'removes a symlink without following it' do + preserve + + expect(Dir.children(preserved)).to eq(['projects']) + expect(tree(outside)).to eq('not-ours.txt' => 'keep me') + end + + it 'still reports the incomplete store copy, without raising, when the directory is already gone' do + FileUtils.remove_entry(config_dir) + + expect(preserve).to match(skipped) + expect(Dir.children(scratch)).to eq(['outside']) # and leaves no empty private directory behind + end + + it 'deletes nothing, and leaves no empty directory behind, when the directory cannot be moved' do + allow(File).to receive(:rename).and_raise(Errno::EACCES, config_dir) + + warning = preserve + + expect(tree(config_dir)).to eq(secrets.merge(transcripts)) + expect(Dir.children(scratch)).to contain_exactly('claude-resume-under-test', 'outside') + expect(warning).to match(skipped) + expect(warning).to include("#{config_dir} could not be moved aside (Permission denied") + end + + # Cut short by something that is not a StandardError (a cancellation, a + # signal), it still has to say what to do: #cleanup leaves the directory + # alone from then on. Where the transcript is, that copies which were not + # removed yet may be left, and which directory to remove once the + # transcript is imported — the private one if the directory was moved + # there, the temp dir itself if it was not. (Its parent would be $TMPDIR.) + context 'when it is interrupted' do + # The whole interrupted notice, for a transcript under +transcript_dir+ + # and +to_remove+ as the directory the user is told to remove. + def interrupted(transcript_dir, to_remove) + to_remove = Regexp.escape(to_remove) + Regexp.new( + 'transcript mirror dropped batches.* Scrubbing was interrupted; the session transcript is under ' \ + "#{Regexp.escape(File.join(transcript_dir, 'projects'))}\\. " \ + "Copies of your credentials and settings .* may be left under #{to_remove} — " \ + "import the transcript into your session store, then remove #{to_remove}\\.$" + ) + end + + it 'names the private directory to remove once the directory was moved there' do + warning = preserve_interrupted_at(File, :unlink) + + expect(tree(preserved).slice(*transcripts.keys)).to eq(transcripts) + expect(File.basename(staging)).to start_with('claude-preserved-resume-') + expect(warning).to match(interrupted(preserved, staging)) + expect(warning).not_to match(/remove #{Regexp.escape(preserved)}\./) + end + + it 'names the temp dir itself when the directory had not been moved yet' do + warning = preserve_interrupted_at(File, :lstat) + + expect(preserved).to eq(config_dir) + expect(Dir.children(scratch)).to contain_exactly('claude-resume-under-test', 'outside') + expect(warning).to match(interrupted(config_dir, config_dir)) + expect(warning).not_to match(/remove #{Regexp.escape(scratch)}\./) + end + end + + # A teardown can run twice. Client#disconnect lets go of its + # MaterializedResume only once preserve_transcripts returned; cut short, a + # later disconnect finds the query handler gone — nothing remembers that + # the mirror dropped batches — and calls #cleanup on the same object. What + # was kept is the only copy of the dropped turns. + context 'when #cleanup is called afterwards' do + it 'leaves the preserved directory alone' do + preserve + + materialized.cleanup + + expect(tree(preserved)).to eq(transcripts) + end + + it 'leaves the transcript in place when the scrub had been interrupted' do + preserve_interrupted_at(File, :unlink) + + materialized.cleanup + + expect(tree(preserved).slice(*transcripts.keys)).to eq(transcripts) + end + + it 'deletes nothing when it had been interrupted before the directory was moved' do + preserve_interrupted_at(File, :lstat) + + materialized.cleanup + + expect(preserved).to eq(config_dir) + expect(tree(config_dir)).to eq(secrets.merge(transcripts)) + end + end + + # An entry that cannot be removed must not pass for scrubbed: backups/ can + # hold the seeded .claude.json, MCP header secrets included. + context 'when an entry resists deletion' do + let(:backups) { File.join(config_dir, 'backups') } + + it 'makes a read-only directory of its own writable and removes it' do + File.chmod(0o500, backups) # listable, not writable: it cannot be moved, its file cannot be unlinked + + warning = preserve + + expect(Dir.children(preserved)).to eq(['projects']) + expect(Dir.children(staging)).to eq(['claude-resume-under-test']) + expect(warning).to match(/Preserving the session transcript/) + expect(warning).not_to match(/Scrubbing failed/) + end + + it 'makes a directory accessible that it could take out but not empty' do + File.chmod(0o600, backups) # writable, so it can be moved; not searchable, so nothing in it can be + + warning = preserve + + expect(Dir.children(preserved)).to eq(['projects']) + expect(Dir.children(staging)).to eq(['claude-resume-under-test']) + expect(warning).not_to match(/Scrubbing failed/) + end + + it 'neither follows nor changes a symlink while it makes a directory writable' do + File.symlink(outside, File.join(backups, 'elsewhere')) + File.chmod(0o555, outside) + File.chmod(0o500, backups) + + preserve + + expect(Dir.children(preserved)).to eq(['projects']) + expect(mode_of(outside)).to eq('555') + expect(tree(outside)).to eq('not-ours.txt' => 'keep me') + end + + it 'does not chmod a directory that another user owns' do + File.chmod(0o500, backups) + allow_any_instance_of(File::Stat).to receive(:owned?).and_return(false) + + warning = preserve + + expect(mode_of(File.join(preserved, 'backups'))).to eq('500') + expect(warning).to match(/Scrubbing failed: could not remove backups \(Operation not permitted/) + end + + it 'says scrubbing failed, and names what is left, when it cannot take an entry out' do + allow(File).to receive(:rename).and_wrap_original do |original, from, to| + raise Errno::EACCES, from if File.basename(from) == 'backups' + + original.call(from, to) + end + + warning = preserve + + expect(Dir.children(preserved)).to contain_exactly('backups', 'projects') + expect(tree(preserved).slice(*transcripts.keys)).to eq(transcripts) + expect(warning).to match(/Preserving the session transcript.* Scrubbing failed: could not remove backups \(Permission denied/) + expect(warning).to match(/What is left is under #{Regexp.escape(staging)}/) + end + + it 'names an entry it had already taken out but could not finish removing' do + taken = nil + allow(File).to receive(:rename).and_wrap_original do |original, from, to| + taken = to if File.basename(from) == 'backups' + original.call(from, to) + end + refusing = true + allow(Dir).to receive(:rmdir).and_wrap_original do |original, path| + raise Errno::EACCES, path if refusing && path == taken + + original.call(path) + end + + warning = preserve + refusing = false + + expect(Dir.children(preserved)).to eq(['projects']) # it is no longer next to the transcript ... + expect(File).to exist(taken) # ... but it is not gone either + expect(warning).to match(/Scrubbing failed: could not remove backups \(Permission denied.*What is left is under #{Regexp.escape(staging)}/) + end + end + + # Something that still holds a directory open can write into it after it + # was moved into the trash. A listing cannot tell: the entry may arrive + # right after it, and the listing may have been empty (on ext4 the empty + # sessions/ was the first directory listed; on APFS it never was). rmdir + # is what notices, and the directory is then emptied again. + listing_orders.each do |how, order| + context "when an entry arrives in a directory that is being emptied (entries listed #{how})" do + # Writes one file into the first directory of the trash whose listing + # the block accepts, right after that listing. + def late_arrival(order) + { arrived: false }.tap do |late| + list_children(order) do |path, names| + next if late[:arrived] || !in_trash?(path) || !yield(names) + + late[:arrived] = true + File.write(File.join(path, 'late.json'), claude_json) + end + end + end + + def expect_nothing_left(warning) + expect(Dir.children(preserved)).to eq(['projects']) + expect(Dir.children(staging)).to eq(['claude-resume-under-test']) + expect(warning).not_to match(/Scrubbing/) + end + + it 'takes it out too, after a listing that had entries' do + late = late_arrival(order, &:any?) + + warning = preserve + + expect(late[:arrived]).to be(true) + expect_nothing_left(warning) + end + + it 'takes it out too, after a listing that was empty' do + late = late_arrival(order, &:empty?) + + warning = preserve + + expect(late[:arrived]).to be(true) + expect_nothing_left(warning) + end + + it 'does the same when rmdir answers EEXIST for a directory that is not empty, as some systems do' do + allow(Dir).to receive(:rmdir).and_wrap_original do |original, path| + raise Errno::EEXIST, path unless Dir.empty?(path) + + original.call(path) + end + late = late_arrival(order, &:empty?) + + warning = preserve + + expect(late[:arrived]).to be(true) + expect_nothing_left(warning) + end + + it 'gives up after EMPTYING_PASSES passes when entries keep arriving, and says what is left' do + kept_busy = nil + arrivals = 0 + list_children(order) do |path, _names| + # The ceiling is for a scrub that would never give up. + next unless in_trash?(path) && (kept_busy ||= path) == path && arrivals < 10 + + File.write(File.join(path, "late-#{arrivals += 1}.json"), claude_json) + end + + warning = preserve + + expect(arrivals).to eq(ClaudeAgentSDK::SessionResume::Scrubber::EMPTYING_PASSES) + expect(warning).to match(/Scrubbing failed: could not remove \S+ \(Directory not empty/) + expect(Dir.children(staging)).to contain_exactly('claude-resume-under-test', a_string_starting_with('scrub-')) + expect(tree(preserved).slice(*transcripts.keys)).to eq(transcripts) + end + end + end + + # Teardown runs on the reactor, and a fiber's stack is small. A scrub that + # recursed once per directory level ran out of it a few hundred levels + # down — one `mkdir -p` makes such a tree — and SystemStackError is not a + # StandardError: it left preserve_transcripts. + context 'when a tree is very deep' do + # A path can only be so long, so the chain is built in pieces of at most + # 150 levels: each new piece takes the chain so far under its deepest + # directory. Returns the top of the chain. + def deep_chain(levels) + chain = nil + levels.fdiv(150).ceil.times do |index| + top = File.join(scratch, "piece-#{index}") + deepest = File.join(top, Array.new([150, levels - (index * 150)].min - 1, 'd')) + FileUtils.mkdir_p(deepest) + chain ? File.rename(chain, File.join(deepest, 'd')) : File.write(File.join(deepest, 'secret.json'), claude_json) + chain = top + end + chain + end + + # The teardown above cannot take such a tree down: unlock and FileUtils + # recurse, and the paths get too long. So whatever a failing example left + # is removed here first, without either: every entry is renamed to the + # top of the scratch directory before it is looked at. + after do + flattened = 0 + pending = Dir.children(scratch).map { |name| File.join(scratch, name) } + until pending.empty? + path = pending.pop + next File.unlink(path) unless File.lstat(path).directory? + + Dir.children(path).each do |child| + pending << File.join(scratch, "flat-#{flattened += 1}") + File.rename(File.join(path, child), pending.last) + end + Dir.rmdir(path) + end + end + + it 'scrubs a chain of 1000 directories on the reactor' do + File.rename(deep_chain(1000), File.join(config_dir, 'deep')) + + warning = Sync { preserve } + + expect(Dir.children(preserved)).to eq(['projects']) + expect(Dir.children(staging)).to eq(['claude-resume-under-test']) # and nothing is left in the trash + expect(tree(preserved)).to eq(transcripts) + expect(warning).not_to match(/Scrubbing/) + end + end + + # A read-only root cannot be moved. Skipping the scrub would leave the + # credential copies behind, so the root is repaired first — through a + # handle that has to be the directory the SDK created. + context 'when the directory itself is read-only' do + it 'makes it writable, moves it and scrubs it' do + File.chmod(0o500, config_dir) + + warning = preserve + + expect(Dir.children(preserved)).to eq(['projects']) + expect(tree(preserved)).to eq(transcripts) + expect(warning).to match(/Preserving the session transcript/) + expect(warning).not_to match(/Scrubbing/) + end + + it 'skips the scrub, deleting nothing, when it cannot be made writable' do + skip 'a superuser can open any directory' if Process.euid.zero? + + File.chmod(0o300, config_dir) # its owner cannot even open it + + warning = preserve + + File.chmod(0o700, config_dir) + expect(tree(config_dir)).to eq(secrets.merge(transcripts)) + expect(Dir.children(scratch)).to contain_exactly('claude-resume-under-test', 'outside') + expect(warning).to match(skipped) + end + end + + # Anything that knows CLAUDE_CONFIG_DIR can replace the temp dir before + # teardown. "Delete every entry but projects/" must then delete nothing, + # rather than the entries of whatever the path leads to now. + # + # The original directory is always renamed aside, never deleted, before + # its replacement appears: a freshly created directory may otherwise reuse + # its inode number and look like the same directory. + context 'when the directory is no longer the one the SDK created' do + let(:unrelated) { File.join(scratch, 'unrelated') } + let(:aside) { File.join(scratch, 'the-sdk-directory-moved-aside') } + + before do + write_tree(unrelated, 'projects/-elsewhere/x.jsonl' => "{}\n", 'innocent/keep.txt' => 'synthetic', + '.claude.json' => '{}') + end + + it 'deletes nothing when the path has become a symlink: it moves the link, not its target' do + File.rename(config_dir, aside) + File.symlink(unrelated, config_dir) + File.chmod(0o555, unrelated) + before_scrub = tree(unrelated) + + warning = preserve + + expect(tree(unrelated)).to eq(before_scrub) + expect(mode_of(unrelated)).to eq('555') + expect(tree(aside)).to eq(secrets.merge(transcripts)) + expect(File.symlink?(preserved)).to be(true) + expect(File.readlink(preserved)).to eq(unrelated) + expect(warning).to match(skipped) + expect(warning).to include(preserved) + end + + it 'deletes nothing in a different directory at the same path' do + File.rename(config_dir, aside) + FileUtils.cp_r(unrelated, config_dir) + before_scrub = tree(config_dir) + + warning = preserve + + expect(tree(preserved)).to eq(before_scrub) + expect(warning).to match(skipped) + end + + it 'does not make a read-only directory writable when it is not the one the SDK created' do + File.rename(config_dir, aside) + FileUtils.cp_r(unrelated, config_dir) + File.chmod(0o500, config_dir) + + warning = preserve + + expect(mode_of(config_dir)).to eq('500') # still where it was, as it was + expect(Dir.children(scratch)).to contain_exactly('claude-resume-under-test', 'outside', 'unrelated', + File.basename(aside)) + expect(warning).to match(skipped) + end + + it 'deletes nothing when it is handed a symlink as the directory' do + link = File.join(scratch, 'replaced-config') + File.symlink(unrelated, link) + handed_a_link = described_class.new(config_dir: link, resume_session_id: sid) + before_scrub = tree(unrelated) + + warning = stderr_of { handed_a_link.preserve_transcripts } + + expect(tree(unrelated)).to eq(before_scrub) + expect(warning).to match(skipped) + end + + # What counts is the directory materialize_resume_session created, not + # whatever is at the path by the time the MaterializedResume is built. + it 'compares with the directory as created, even if it was replaced during materialization' do + created = [] + allow(Dir).to receive(:mktmpdir).and_wrap_original do |original, *args, **options, &block| + next original.call(*args, **options, &block) unless args == ['claude-resume-'] + + original.call('claude-resume-', scratch).tap { |dir| created << dir } + end + replace_temp_dir = lambda do + File.rename(created.last, aside) + FileUtils.cp_r(unrelated, created.last) + end + store = Class.new(ClaudeAgentSDK::InMemorySessionStore) do + define_method(:list_subkeys) do |_key| + replace_temp_dir.call # runs after the transcript and the seed files are written + [] + end + end.new + project_dir = File.join(scratch, 'project') + caller_config_dir = File.join(scratch, 'caller-config') # empty: nothing seeded from the developer's own + [project_dir, caller_config_dir].each { |dir| FileUtils.mkdir_p(dir) } + store.append({ 'project_key' => ClaudeAgentSDK.project_key_for_directory(project_dir), 'session_id' => sid }, + [{ 'type' => 'user', 'uuid' => 'u1', 'sessionId' => sid, 'message' => { 'role' => 'user', 'content' => 'hi' } }]) + + from_the_store = ClaudeAgentSDK::SessionResume.materialize_resume_session( + ClaudeAgentSDK::ClaudeAgentOptions.new(session_store: store, resume: sid, cwd: project_dir, + env: { 'CLAUDE_CONFIG_DIR' => caller_config_dir }) + ) + before_scrub = tree(from_the_store.config_dir) + + warning = stderr_of { from_the_store.preserve_transcripts } + + expect(tree(from_the_store.config_dir)).to eq(before_scrub) + expect(warning).to match(skipped) + end + end + + # The same, with the swap timed against the cleanup itself. Each example + # puts it between two steps of preserve_transcripts by wrapping the call + # that comes next; nothing depends on a clock or on another thread. + context 'when something is swapped while the cleanup runs' do + let(:unrelated) { File.join(scratch, 'unrelated') } + let(:aside) { File.join(scratch, 'the-sdk-directory-moved-aside') } + let(:backups) { File.join(config_dir, 'backups') } + + before do + # An entry for every name the scrub lists, whatever order it walks + # them in, and for what is inside backups/: each one deleted through + # a swapped path would be a file the SDK never wrote. + names = Dir.children(config_dir) - %w[projects linked-elsewhere] + write_tree(unrelated, names.to_h { |name| [name == 'backups' ? 'backups/kept.txt' : name, 'synthetic'] } + .merge('.claude.json.backup.1790881328533' => 'synthetic')) + end + + it 'deletes nothing when the directory becomes a symlink right before it is moved' do + swapped = false + allow(File).to receive(:rename).and_wrap_original do |original, from, to| + if from == config_dir && !swapped + swapped = true + original.call(config_dir, aside) + File.symlink(unrelated, config_dir) + end + original.call(from, to) + end + before_scrub = tree(unrelated) + + warning = preserve + + expect(swapped).to be(true) + expect(tree(unrelated)).to eq(before_scrub) + expect(tree(aside)).to eq(secrets.merge(transcripts)) + expect(warning).to match(skipped) + end + + it 'never follows a symlink planted at the old path once the directory was moved' do + planted = false + allow(File).to receive(:rename).and_wrap_original do |original, from, to| + if File.basename(File.dirname(to)).start_with?('scrub-') && !planted # right before the first removal + planted = true + File.symlink(unrelated, config_dir) + end + original.call(from, to) + end + before_scrub = tree(unrelated) + + warning = preserve + + expect(planted).to be(true) + expect(tree(unrelated)).to eq(before_scrub) + expect(Dir.children(preserved)).to eq(['projects']) + expect(tree(preserved)).to eq(transcripts) + expect(warning).to match(/Preserving the session transcript under #{Regexp.escape(File.join(preserved, 'projects'))}/) + expect(warning).not_to match(/Scrubbing/) + end + + it 'unlinks, and does not follow, an entry that becomes a symlink right before it is removed' do + swapped = false + allow(File).to receive(:rename).and_wrap_original do |original, from, to| + if File.basename(from) == 'backups' && !swapped + swapped = true + original.call(from, "#{from}-aside") + File.symlink(unrelated, from) + end + original.call(from, to) + end + before_scrub = tree(unrelated) + + warning = preserve + + expect(swapped).to be(true) + expect(tree(unrelated)).to eq(before_scrub) + expect(tree(preserved).slice(*transcripts.keys)).to eq(transcripts) + # The real backups/ is still there under its new name, and the warning says so. + expect(Dir.children(preserved)).to contain_exactly('backups-aside', 'projects') + expect(warning).to match(/Scrubbing failed: could not remove backups-aside/) + end + + # A read-only backups/ has to be made writable. Between the look at it + # and the chmod it is swapped for a symlink to a directory elsewhere: + # that directory must keep its mode and its files, and the scrub stops. + listing_orders.each do |how, order| + context "with a read-only backups/ (entries listed #{how})" do + # The names the scrub got for the directory itself, in the order it + # walked them: that it stopped at backups/ shows in what came after. + let(:listed) { [] } + + before do + File.chmod(0o500, backups) + File.chmod(0o555, outside) + list_children(order) do |path, names| + listed.replace(names) if listed.empty? && File.basename(path) == 'claude-resume-under-test' + end + end + + # Gone: what was listed before backups/. Still there: what was listed + # after it, projects/, the symlink the swap put at backups and the real + # directory under the name the swap gave it. + def expect_the_scrub_stopped_at_backups + untouched = listed.drop(listed.index('backups') + 1) + expect(Dir.children(preserved)).to match_array((untouched + %w[projects backups backups-aside]).uniq) + end + + it 'leaves an outside directory alone when backups/ is swapped between the lstat and the chmod' do + swapped = false + allow(File).to receive(:lstat).and_wrap_original do |original, path| + stat = original.call(path) + if !swapped && File.basename(path) == 'backups' && stat.directory? && stat.mode.nobits?(0o200) + swapped = true + File.rename(path, "#{path}-aside") + File.symlink(outside, path) + end + stat + end + + warning = preserve + + expect(swapped).to be(true) + expect(mode_of(outside)).to eq('555') + expect(tree(outside)).to eq('not-ours.txt' => 'keep me') + expect(tree(preserved).slice(*transcripts.keys)).to eq(transcripts) + expect(warning).to match(/Preserving the session transcript.* Scrubbing failed: could not remove .*backups/) + expect(warning).to match(/backups became a symlink/) + expect_the_scrub_stopped_at_backups + end + + it 'stops, leaving an outside directory alone, when backups/ is swapped right after the chmod' do + swapped = false + allow(File).to receive(:open).and_wrap_original do |original, *args, **options, &block| + result = original.call(*args, **options, &block) + if !swapped && File.basename(args.first.to_s) == 'backups' + swapped = true + File.rename(args.first, "#{args.first}-aside") + File.symlink(outside, args.first) + end + result + end + + warning = preserve + + expect(swapped).to be(true) + expect(mode_of(outside)).to eq('555') + expect(tree(outside)).to eq('not-ours.txt' => 'keep me') + expect(tree(preserved).slice(*transcripts.keys)).to eq(transcripts) + expect(warning).to match(/Scrubbing failed: could not remove .*backups/) + expect(warning).to match(/backups was replaced while it was being made writable/) + expect_the_scrub_stopped_at_backups + end + end + end + end + end +end diff --git a/spec/unit/session_resume_keychain_spec.rb b/spec/unit/session_resume_keychain_spec.rb new file mode 100644 index 00000000..dddf78fe --- /dev/null +++ b/spec/unit/session_resume_keychain_spec.rb @@ -0,0 +1,397 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'securerandom' +require 'tmpdir' +require 'json' +require 'fileutils' +require 'open3' +require 'rbconfig' + +# The macOS Keychain bridge of store-backed resume. On macOS the CLI keeps its +# OAuth credentials in the Keychain rather than in .credentials.json, so +# materialization copies them (minus the refresh token) into the temp config +# dir the resumed subprocess runs against. +# +# No example here runs the real `security` binary or reads a real Keychain. +# The bridge is reached through materialize_resume_session — a stub of +# read_keychain_credentials would hide a wiring regression — and the layer +# below it, capture_with_timeout, is replaced by a recorder that answers like +# `security find-generic-password -w` does. capture_with_timeout itself is +# exercised against a real child process that is not `security`. +# +# `keychain: true` opts out of the suite-wide default stub of +# read_keychain_credentials (spec/support/keychain_bridge_stub.rb). +RSpec.describe ClaudeAgentSDK::SessionResume, keychain: true do + let(:store) { ClaudeAgentSDK::InMemorySessionStore.new } + let(:cwd) { Dir.mktmpdir } + let(:home) { Dir.mktmpdir } + let(:sid) { SecureRandom.uuid } + let(:security_calls) { [] } + let(:materializations) { [] } + + # What the CLI stores in the Keychain item (values are fake). + let(:oauth) do + { 'accessToken' => 'fake-access-token', 'refreshToken' => 'fake-refresh-token', + 'expiresAt' => 1_790_000_000_000, 'scopes' => %w[user:inference user:profile], + 'subscriptionType' => 'max', 'rateLimitTier' => 'default_claude_max_20x' } + end + let(:keychain_payload) { JSON.generate('claudeAiOauth' => oauth) } + + # The bridge keys off the environment the CHILD will see, which falls back to + # the parent's for every key options.env does not set. Pin the ones it reads, + # so the examples behave the same on any machine: a shell or CI job exporting + # ANTHROPIC_API_KEY would otherwise skip the bridge altogether. + around do |example| + names = %w[CLAUDE_CONFIG_DIR ANTHROPIC_API_KEY CLAUDE_CODE_OAUTH_TOKEN USER] + saved = names.to_h { |name| [name, ENV.fetch(name, nil)] } + names.each { |name| ENV.delete(name) } + ENV['USER'] = 'spec-user' + example.run + ensure + saved.each { |name, value| value.nil? ? ENV.delete(name) : (ENV[name] = value) } + end + + before do + stub_host_os('darwin24') # the bridge is a no-op elsewhere, and CI runs on Linux + store.append({ 'project_key' => ClaudeAgentSDK.project_key_for_directory(cwd), 'session_id' => sid }, + [{ 'parentUuid' => nil, 'isSidechain' => false, 'userType' => 'external', 'cwd' => cwd, + 'sessionId' => sid, 'version' => '2.1.286', 'gitBranch' => 'main', 'type' => 'user', + 'message' => { 'role' => 'user', 'content' => 'hello' }, 'uuid' => SecureRandom.uuid, + 'timestamp' => '2026-09-01T12:00:00.000Z' }]) + end + + after do + materializations.each(&:cleanup) + [cwd, home].each { |dir| FileUtils.remove_entry(dir) if File.directory?(dir) } + end + + def stub_host_os(host_os) + allow(RbConfig::CONFIG).to receive(:[]).and_call_original + allow(RbConfig::CONFIG).to receive(:[]).with('host_os').and_return(host_os) + end + + # Stand-in for `security find-generic-password -w`, which prints the secret + # and a newline and exits 0 on a hit, and prints nothing and exits non-zero + # when the item does not exist. +exit_ok+ nil is capture_with_timeout's own + # answer for a command that timed out or could not be run: [nil, nil]. + def stub_security(stdout, exit_ok) + status = exit_ok.nil? ? nil : instance_double(Process::Status, success?: exit_ok) + allow(described_class).to receive(:capture_with_timeout) do |argv, timeout_s| + security_calls << [argv, timeout_s] + [stdout, status] + end + end + + # Store-backed resume as query() and Client#connect run it. HOME is a + # scratch directory, so nothing is seeded from the developer's own ~/.claude. + def materialize(env = {}) + options = ClaudeAgentSDK::ClaudeAgentOptions.new(session_store: store, resume: sid, cwd: cwd, + env: { 'HOME' => home }.merge(env)) + described_class.materialize_resume_session(options).tap { |mat| materializations << mat } + end + + def seeded_credentials(materialized) + File.join(materialized.config_dir, '.credentials.json') + end + + def write_credentials_file(config_dir, access_token) + FileUtils.mkdir_p(config_dir) + File.write(File.join(config_dir, '.credentials.json'), + JSON.generate('claudeAiOauth' => oauth.merge('accessToken' => access_token))) + end + + def security_argv(service) + ['security', 'find-generic-password', '-a', 'spec-user', '-w', '-s', service] + end + + describe 'with the default config dir' do + it "reads the CLI's default Keychain entry and seeds it without the refresh token, owner-only" do + stub_security("#{keychain_payload}\n", true) + + materialized = materialize + + expect(security_calls).to eq([[security_argv('Claude Code-credentials'), described_class::KEYCHAIN_TIMEOUT_SECONDS]]) + seeded = seeded_credentials(materialized) + expect(JSON.parse(File.read(seeded))).to eq('claudeAiOauth' => oauth.except('refreshToken')) + expect(File.read(seeded)).not_to include('fake-refresh-token') + expect(format('%o', File.stat(seeded).mode & 0o777)).to eq('600') + end + + it 'lets a Keychain hit override ~/.claude/.credentials.json' do + write_credentials_file(File.join(home, '.claude'), 'from-the-file') + stub_security("#{keychain_payload}\n", true) + + materialized = materialize + + expect(JSON.parse(File.read(seeded_credentials(materialized))).dig('claudeAiOauth', 'accessToken')) + .to eq('fake-access-token') + end + + it 'falls back to ~/.claude/.credentials.json when the Keychain has no entry' do + write_credentials_file(File.join(home, '.claude'), 'from-the-file') + stub_security('', false) + + materialized = materialize + + expect(security_calls.length).to eq(1) + expect(JSON.parse(File.read(seeded_credentials(materialized)))) + .to eq('claudeAiOauth' => oauth.except('refreshToken').merge('accessToken' => 'from-the-file')) + end + + { + 'the lookup exits non-zero (no such entry)' => ['', false], + 'the lookup exits non-zero after printing something' => [%({"claudeAiOauth":{"accessToken":"untrusted"}}\n), false], + 'the entry is empty' => ["\n", true], + 'the lookup times out or cannot be run' => [nil, nil] + }.each do |outcome, (stdout, exit_ok)| + it "seeds no credentials file when #{outcome}, and still materializes the transcript" do + stub_security(stdout, exit_ok) + + materialized = materialize + + expect(security_calls.length).to eq(1) + expect(File).not_to exist(seeded_credentials(materialized)) + expect(Dir.glob(File.join(materialized.config_dir, 'projects', '*', "#{sid}.jsonl")).length).to eq(1) + end + end + + %w[ANTHROPIC_API_KEY CLAUDE_CODE_OAUTH_TOKEN].each do |name| + it "leaves the Keychain alone when options.env carries #{name}" do + stub_security("#{keychain_payload}\n", true) + + materialized = materialize(name => 'from-the-env') + + expect(security_calls).to be_empty + expect(File).not_to exist(seeded_credentials(materialized)) + end + + it "leaves the Keychain alone when the child inherits #{name} from the parent" do + stub_security("#{keychain_payload}\n", true) + ENV[name] = 'from-the-env' + + materialize + + expect(security_calls).to be_empty + end + end + + it 'leaves the Keychain alone off macOS' do + stub_host_os('linux-gnu') + stub_security("#{keychain_payload}\n", true) + + materialized = materialize + + expect(security_calls).to be_empty + expect(File).not_to exist(seeded_credentials(materialized)) + end + + it 'treats an empty CLAUDE_CONFIG_DIR as the default config dir, like the CLI' do + stub_security("#{keychain_payload}\n", true) + + materialize('CLAUDE_CONFIG_DIR' => '') + + expect(security_calls.map(&:first)).to eq([security_argv('Claude Code-credentials')]) + end + end + + # With a custom CLAUDE_CONFIG_DIR the CLI keeps its entry under + # "Claude Code-credentials-", hashing the + # directory as the environment gives it, NFC-normalized — and writes no + # .credentials.json there. The expected suffixes below are literals computed + # outside Ruby (`printf '%s' | shasum -a 256 | cut -c1-8`), so they pin + # the derivation instead of restating it. + describe 'with a custom CLAUDE_CONFIG_DIR' do + # No such directory: nothing to copy from it, and no .credentials.json. + let(:custom_dir) { '/nonexistent/claude-profiles/work' } + let(:custom_service) { 'Claude Code-credentials-43da4a82' } + + it 'reads the entry the CLI keeps for that directory and seeds it without the refresh token' do + stub_security("#{keychain_payload}\n", true) + + materialized = materialize('CLAUDE_CONFIG_DIR' => custom_dir) + + expect(security_calls).to eq([[security_argv(custom_service), described_class::KEYCHAIN_TIMEOUT_SECONDS]]) + seeded = seeded_credentials(materialized) + expect(JSON.parse(File.read(seeded))).to eq('claudeAiOauth' => oauth.except('refreshToken')) + expect(format('%o', File.stat(seeded).mode & 0o777)).to eq('600') + end + + it 'reads it for a config dir the child inherits from the parent environment' do + stub_security("#{keychain_payload}\n", true) + ENV['CLAUDE_CONFIG_DIR'] = custom_dir + + materialized = materialize + + expect(security_calls.map(&:first)).to eq([security_argv(custom_service)]) + expect(File).to exist(seeded_credentials(materialized)) + end + + # "cafe" + U+0301 COMBINING ACUTE ACCENT: the decomposed spelling of the + # composed one-code-point form (U+00E9) that the CLI hashes. Built from + # the code point so no editor can quietly re-compose the literal. 0e56012f + # is the digest of the composed spelling; the bytes as given hash to + # d3d97122. + context 'when the directory is given in decomposed (NFD) form' do + let(:custom_dir) { "/nonexistent/claude-profiles/cafe#{[0x301].pack('U')}" } + let(:custom_service) { 'Claude Code-credentials-0e56012f' } + + it 'hashes its NFC form' do + stub_security("#{keychain_payload}\n", true) + + materialize('CLAUDE_CONFIG_DIR' => custom_dir) + + expect(custom_dir.unicode_normalize(:nfc)).not_to eq(custom_dir) + expect(security_calls.map(&:first)).to eq([security_argv(custom_service)]) + end + + it 'hashes the same form when Ruby tagged it as binary (a non-ASCII ENV value under a C locale)' do + stub_security("#{keychain_payload}\n", true) + + materialize('CLAUDE_CONFIG_DIR' => custom_dir.b) + + expect(security_calls.map(&:first)).to eq([security_argv(custom_service)]) + end + end + + it 'seeds nothing when the Keychain has no entry for that directory' do + stub_security('', false) + + materialized = materialize('CLAUDE_CONFIG_DIR' => custom_dir) + + expect(security_calls.length).to eq(1) + expect(File).not_to exist(seeded_credentials(materialized)) + end + + # Which source the CLI prefers when a custom config dir has both is not + # established, so the file keeps winning there, as it always has. + context 'when that directory has a .credentials.json' do + let(:custom_dir) { Dir.mktmpdir } + + after { FileUtils.remove_entry(custom_dir) if File.directory?(custom_dir) } + + it 'seeds the file and leaves the Keychain alone' do + write_credentials_file(custom_dir, 'from-the-file') + stub_security("#{keychain_payload}\n", true) + + materialized = materialize('CLAUDE_CONFIG_DIR' => custom_dir) + + expect(security_calls).to be_empty + expect(JSON.parse(File.read(seeded_credentials(materialized)))) + .to eq('claudeAiOauth' => oauth.except('refreshToken').merge('accessToken' => 'from-the-file')) + end + + it 'leaves the Keychain alone even when what is there cannot be read as a file' do + Dir.mkdir(File.join(custom_dir, '.credentials.json')) + stub_security("#{keychain_payload}\n", true) + + materialized = nil + expect { materialized = materialize('CLAUDE_CONFIG_DIR' => custom_dir) } + .to output(/skipping .*\.credentials\.json \(not a regular file/).to_stderr + + expect(security_calls).to be_empty + expect(File).not_to exist(seeded_credentials(materialized)) + end + end + + %w[ANTHROPIC_API_KEY CLAUDE_CODE_OAUTH_TOKEN].each do |name| + it "leaves the Keychain alone when options.env carries #{name}" do + stub_security("#{keychain_payload}\n", true) + + materialized = materialize('CLAUDE_CONFIG_DIR' => custom_dir, name => 'from-the-env') + + expect(security_calls).to be_empty + expect(File).not_to exist(seeded_credentials(materialized)) + end + end + + it 'leaves the Keychain alone off macOS' do + stub_host_os('linux-gnu') + stub_security("#{keychain_payload}\n", true) + + materialized = materialize('CLAUDE_CONFIG_DIR' => custom_dir) + + expect(security_calls).to be_empty + expect(File).not_to exist(seeded_credentials(materialized)) + end + end + + describe '.capture_with_timeout' do + # A real child process (never `security`). A child that is expected to exit + # gets a deadline it cannot plausibly miss — the call returns the moment it + # exits — so machine load never decides the outcome; only the child that + # never exits meets a short one. + let(:ample) { 120 } + + def ruby_child(script) + [RbConfig.ruby, '--disable-gems', '-e', script] + end + + def capture(argv, timeout_s) + described_class.send(:capture_with_timeout, argv, timeout_s) + end + + it "returns the child's stdout and a successful status" do + stdout, status = capture(ruby_child('print "from the child"'), ample) + + expect(stdout).to eq('from the child') + expect(status).to be_success + end + + it 'returns everything a child wrote, past the size of a pipe buffer' do + stdout, status = capture(ruby_child('print "x" * 300_000'), ample) + + expect(stdout.bytesize).to eq(300_000) + expect(status).to be_success + end + + it 'returns the output and the failed status of a child that exits non-zero' do + stdout, status = capture(ruby_child('print "partial"; exit 3'), ample) + + expect(stdout).to eq('partial') + expect(status.exitstatus).to eq(3) + expect(status).not_to be_success + end + + it 'kills and reaps a child that outlives the deadline, returning [nil, nil]' do + waiter = nil + allow(Open3).to receive(:popen3).and_wrap_original do |original, *argv| + streams_and_waiter = original.call(*argv) + waiter = streams_and_waiter.last + streams_and_waiter + end + + expect(capture(ruby_child('sleep'), 0.05)).to eq([nil, nil]) + expect(waiter).not_to be_alive # already reaped: no zombie, no stray `security` + expect(waiter.value.termsig).to eq(Signal.list['KILL']) + end + + it 'returns [nil, nil] for a command that cannot be spawned' do + expect(capture([File.join(home, 'no-such-command')], ample)).to eq([nil, nil]) + end + end + + # spec/support/keychain_bridge_stub.rb switches the bridge off for every + # example in the suite that does not carry `keychain: true`. This group gives + # the tag up again, to pin that default exactly where the bridge would + # otherwise run: macOS, no env auth, and a config dir it covers. + describe 'the suite-wide default, without keychain: true', keychain: false do + it 'keeps materialization away from the lookup' do + stub_security("#{keychain_payload}\n", true) + + materialized = materialize + + expect(security_calls).to be_empty + expect(File).not_to exist(seeded_credentials(materialized)) + end + + it 'does so for a custom config dir without a .credentials.json as well' do + stub_security("#{keychain_payload}\n", true) + + materialized = materialize('CLAUDE_CONFIG_DIR' => '/nonexistent/claude-profiles/work') + + expect(security_calls).to be_empty + expect(File).not_to exist(seeded_credentials(materialized)) + end + end +end diff --git a/spec/unit/session_resume_nil_options_spec.rb b/spec/unit/session_resume_nil_options_spec.rb new file mode 100644 index 00000000..85eef26a --- /dev/null +++ b/spec/unit/session_resume_nil_options_spec.rb @@ -0,0 +1,86 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'securerandom' +require 'tmpdir' +require 'fileutils' + +# ClaudeAgentOptions fills `env` and `load_timeout_ms` with their defaults in +# the constructor only. Both attributes are nilable, and the natural ways to +# clear one afterwards — `options.dup_with(env: nil)`, `options.env = nil` — +# leave nil in place. Store-backed resume has to read them as their defaults +# (60 s; no overrides) rather than raise NoMethodError on nil. +RSpec.describe ClaudeAgentSDK::SessionResume do + let(:store) { ClaudeAgentSDK::InMemorySessionStore.new } + let(:cwd) { Dir.mktmpdir } + let(:caller_config_dir) { Dir.mktmpdir } + let(:sid) { SecureRandom.uuid } + let(:materializations) { [] } + let(:options) { ClaudeAgentSDK::ClaudeAgentOptions.new(session_store: store, resume: sid, cwd: cwd) } + + # Without options.env the child's config dir is the parent's: point that at + # a scratch directory, so nothing is seeded from the developer's own. + around do |example| + previous = ENV.fetch('CLAUDE_CONFIG_DIR', nil) + ENV['CLAUDE_CONFIG_DIR'] = caller_config_dir + example.run + ensure + previous.nil? ? ENV.delete('CLAUDE_CONFIG_DIR') : (ENV['CLAUDE_CONFIG_DIR'] = previous) + end + + before do + store.append({ 'project_key' => ClaudeAgentSDK.project_key_for_directory(cwd), 'session_id' => sid }, + [{ 'type' => 'user', 'uuid' => SecureRandom.uuid, 'sessionId' => sid, + 'message' => { 'role' => 'user', 'content' => 'hello' } }]) + end + + after do + materializations.each(&:cleanup) + [cwd, caller_config_dir].each { |dir| FileUtils.remove_entry(dir) if File.directory?(dir) } + end + + def materialize(options) + described_class.materialize_resume_session(options).tap { |materialized| materializations << materialized if materialized } + end + + it 'bounds the store calls by the 60 s default when load_timeout_ms is nil' do + allow(ClaudeAgentSDK::FiberBoundary).to receive(:invoke).and_call_original + + materialized = materialize(options.dup_with(load_timeout_ms: nil)) + + expect(materialized.resume_session_id).to eq(sid) + expect(ClaudeAgentSDK::FiberBoundary).to have_received(:invoke).with(hash_including(timeout: 60.0)).at_least(:once) + end + + it 'still honors an explicit load_timeout_ms of 0' do + allow(ClaudeAgentSDK::FiberBoundary).to receive(:invoke).and_return([]) + + materialize(options.dup_with(load_timeout_ms: 0)) + + expect(ClaudeAgentSDK::FiberBoundary).to have_received(:invoke).with(hash_including(timeout: 0.0)).at_least(:once) + end + + it 'materializes from the parent environment and repoints the options when env is nil' do + File.write(File.join(caller_config_dir, 'settings.json'), '{"apiKeyHelper":"/bin/print-key"}') + without_env = options.dup_with(env: nil) + + materialized = materialize(without_env) + applied = described_class.apply_materialized_options(without_env, materialized) + + expect(File.read(File.join(materialized.config_dir, 'settings.json'))).to eq('{"apiKeyHelper":"/bin/print-key"}') + expect(applied.env).to eq('CLAUDE_CONFIG_DIR' => materialized.config_dir) + expect(applied.resume).to eq(sid) + expect(without_env.env).to be_nil # the caller's options are left as they were + end + + it 'gets Client past materialization with both set back to nil' do + cleared = options.dup_with(env: nil, load_timeout_ms: nil) + client = ClaudeAgentSDK::Client.new(options: cleared) + + applied = client.send(:materialize_resume, cleared) + materializations << client.instance_variable_get(:@materialized) + + expect(applied.env).to eq('CLAUDE_CONFIG_DIR' => materializations.first.config_dir) + expect(applied.resume).to eq(sid) + end +end diff --git a/spec/unit/session_resume_spec.rb b/spec/unit/session_resume_spec.rb index f30c1228..a4ed5783 100644 --- a/spec/unit/session_resume_spec.rb +++ b/spec/unit/session_resume_spec.rb @@ -1049,10 +1049,10 @@ def mode_of(path) # not delete it. It keeps the transcripts but scrubs the credential copies. describe 'MaterializedResume#preserve_transcripts' do it 'removes credential copies, keeps transcripts, warns, and never rmtrees' do - Dir.mktmpdir do |dir| - transcript = File.join(dir, 'projects', 'pk', 'sid.jsonl') - FileUtils.mkdir_p(File.dirname(transcript)) - File.write(transcript, "{}\n") + Dir.mktmpdir do |scratch| + dir = File.join(scratch, 'claude-resume-spec') + FileUtils.mkdir_p(File.join(dir, 'projects', 'pk')) + File.write(File.join(dir, 'projects', 'pk', 'sid.jsonl'), "{}\n") # settings.json / cowork_settings.json are seeded from the caller's # config dir too, and their env blocks routinely carry API keys — the # scrub must cover them, not just the credential files. @@ -1063,10 +1063,14 @@ def mode_of(path) materialized = ClaudeAgentSDK::MaterializedResume.new(config_dir: dir, resume_session_id: 'sid') expect { materialized.preserve_transcripts }.to output(/[Pp]reserving/).to_stderr + # The directory is kept in a private one created next to it (so still + # under scratch): that is where the transcript is now. + preserved = materialized.config_dir + expect(File.dirname(preserved, 2)).to eq(scratch) ['.credentials.json', '.claude.json', 'settings.json', 'cowork_settings.json'].each do |name| - expect(File).not_to exist(File.join(dir, name)) + expect(File).not_to exist(File.join(preserved, name)) end - expect(File).to exist(transcript) + expect(File).to exist(File.join(preserved, 'projects', 'pk', 'sid.jsonl')) end end end diff --git a/spec/unit/transcript_mirror_batcher_backoff_spec.rb b/spec/unit/transcript_mirror_batcher_backoff_spec.rb new file mode 100644 index 00000000..5f5a4588 --- /dev/null +++ b/spec/unit/transcript_mirror_batcher_backoff_spec.rb @@ -0,0 +1,77 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'async' + +RSpec.describe ClaudeAgentSDK::TranscriptMirrorBatcher do + # Between attempts the batcher slept with Async::Task#sleep when a reactor + # was running. async deprecated that in favor of Kernel#sleep, which is + # scheduler-aware, and warns about it when warnings are on (`ruby -w`): + # one line on stderr per retry. + describe 'retry backoff' do + let(:projects) { '/cfg/projects' } + let(:file_path) { "#{projects}/-Users-dev-app/11111111-1111-4111-8111-111111111111.jsonl" } + + # What `ruby -w` turns on. + around do |example| + verbose = $VERBOSE + deprecated = Warning[:deprecated] + $VERBOSE = true + Warning[:deprecated] = true + example.run + ensure + $VERBOSE = verbose + Warning[:deprecated] = deprecated + end + + it 'retries inside a reactor without a deprecation warning' do + stub_const("#{described_class}::MIRROR_APPEND_BACKOFF_S", [0, 0].freeze) # nothing to wait for here + attempts = 0 + store = Class.new(ClaudeAgentSDK::SessionStore) do + define_method(:append) do |_key, _entries| + attempts += 1 + raise IOError, 'connection reset' if attempts == 1 + end + + def load(_key) = nil + end.new + batcher = described_class.new(store: store, projects_dir: projects, on_error: ->(_key, _message) {}) + + expect do + Async do + batcher.enqueue(file_path, [{ type: 'user', uuid: 'a' }]) + batcher.flush + end.wait + end.not_to output(/Async::Task#sleep/).to_stderr + + expect(attempts).to eq(2) # the backoff ran: one failure, one retry + end + + # With [0, 0] intervals the example above cannot tell a backoff from no + # backoff at all. Here the configured intervals have to reach + # Kernel#sleep — intercepted on the batcher, so nothing actually waits. + it 'sleeps the configured interval before each retry' do + stub_const("#{described_class}::MIRROR_APPEND_BACKOFF_S", [0.25, 0.5].freeze) + attempts = 0 + store = Class.new(ClaudeAgentSDK::SessionStore) do + define_method(:append) do |_key, _entries| + attempts += 1 + raise IOError, 'connection reset' if attempts < 3 + end + + def load(_key) = nil + end.new + batcher = described_class.new(store: store, projects_dir: projects, on_error: ->(_key, _message) {}) + sleeps = [] + allow(batcher).to receive(:sleep) { |seconds| sleeps << seconds } + + Async do + batcher.enqueue(file_path, [{ type: 'user', uuid: 'a' }]) + batcher.flush + end.wait + + expect(attempts).to eq(3) + expect(sleeps).to eq([0.25, 0.5]) + end + end +end diff --git a/spec/unit/transcript_mirror_batcher_cancellation_spec.rb b/spec/unit/transcript_mirror_batcher_cancellation_spec.rb new file mode 100644 index 00000000..a2ea58dc --- /dev/null +++ b/spec/unit/transcript_mirror_batcher_cancellation_spec.rb @@ -0,0 +1,168 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'async' +require 'async/condition' + +# A drain takes the pending buffer only once it holds the lock. It used to +# detach first and then queue for the lock, so a #flush cancelled while it +# waited behind an in-flight append (a task stopped, a timeout around the +# caller) took its batch down with it: counted by #batches_dropped?, but never +# appended and never reported through on_error (upstream Python PR #1289). +# +# Every interleaving below is forced with a gate the example controls — a +# Thread::Queue for the default thread-hop store, an Async::Condition for a +# fiber-native one. Nothing waits on a clock. +RSpec.describe ClaudeAgentSDK::TranscriptMirrorBatcher do + let(:projects) { '/cfg/projects' } + let(:file_path) { "#{projects}/-Users-dev-app/11111111-1111-4111-8111-111111111111.jsonl" } + let(:errors) { [] } + let(:on_error) { ->(key, message) { errors << [key, message] } } + + # Each append announces itself on `started`, then parks its (batcher-spawned) + # worker thread on `gate` until the example closes it. + let(:gated_store) do + Class.new(ClaudeAgentSDK::SessionStore) do + attr_reader :started, :gate + + def initialize + super + @entries = [] + @mutex = Mutex.new + @started = Thread::Queue.new + @gate = Thread::Queue.new + end + + def append(_key, entries) + @started.push(true) + @gate.pop # nil once the gate is closed + @mutex.synchronize { @entries.concat(entries) } + end + + def load(_key) = @mutex.synchronize { @entries.dup } + def uuids = load(nil).map { |entry| entry['uuid'] } + end.new + end + + # Fiber-native adapter: #append runs in place on the reactor and parks on a + # condition until the example releases it. + let(:inline_store) do + Class.new(ClaudeAgentSDK::SessionStore) do + attr_reader :uuids + + def initialize + super + @uuids = [] + @entered = false + @released = false + @gate = Async::Condition.new + end + + def callback_scheduling = :inline + def entered? = @entered + def load(_key) = nil + + def append(_key, entries) + @entered = true + @gate.wait until @released + @uuids.concat(entries.map { |entry| entry['uuid'] }) + end + + def release! + @released = true + @gate.signal + end + end.new + end + + def batcher(store) + described_class.new(store: store, projects_dir: projects, on_error: on_error) + end + + # A frame as the transport delivers it: symbol keys. + def frame(uuid) + [{ type: 'user', uuid: uuid, message: { role: 'user', content: uuid } }] + end + + it 'keeps the batch of a flush cancelled while it waits behind an in-flight append' do + b = batcher(gated_store) + Async do |task| + task.with_timeout(30) do + b.enqueue(file_path, frame('one')) + first = task.async { b.flush } # holds the lock; its append is parked on the gate + gated_store.started.pop + b.enqueue(file_path, frame('two')) + waiter = task.async { b.flush } # queued on the lock behind `first` + waiter.stop # ... and cancelled there + b.enqueue(file_path, frame('three')) + + gated_store.gate.close + first.wait + b.close + end + end.wait + + expect(gated_store.uuids).to eq(%w[one two three]) + expect(b.batches_dropped?).to be(false) + expect(errors).to be_empty + end + + it 'keeps it for a fiber-native (inline) store too' do + b = batcher(inline_store) + Async do |task| + task.with_timeout(30) do + b.enqueue(file_path, frame('one')) + first = task.async { b.flush } # in place on the reactor, parked on the condition + expect(inline_store).to be_entered + b.enqueue(file_path, frame('two')) + waiter = task.async { b.flush } + waiter.stop + b.enqueue(file_path, frame('three')) + + inline_store.release! + first.wait + b.close + end + end.wait + + expect(inline_store.uuids).to eq(%w[one two three]) + expect(b.batches_dropped?).to be(false) + expect(errors).to be_empty + end + + # What a drain HAS detached is still lost with it, and counted: the + # materialized resume dir then holds the only copy and must be kept. + it 'still counts a batch as dropped when its drain is cancelled inside the append' do + b = batcher(gated_store) + Async do |task| + task.with_timeout(30) do + b.enqueue(file_path, frame('one')) + flusher = task.async { b.flush } + gated_store.started.pop # the append is in flight: `one` is detached + flusher.stop + gated_store.gate.close + end + end.wait + + expect(b.batches_dropped?).to be(true) + end + + # #close does not chase frames that arrive after it detached. With nothing + # left to drain them they stay buffered, and that counts as a drop. + it 'counts a frame that arrives during the final append as dropped when nothing drains it' do + b = batcher(gated_store) + Async do |task| + task.with_timeout(30) do + b.enqueue(file_path, frame('a')) + closer = task.async { b.close } # detaches `a` under the lock; its append is parked on the gate + gated_store.started.pop + b.enqueue(file_path, frame('late')) # below the thresholds: no drain is scheduled for it + gated_store.gate.close + closer.wait + end + end.wait + + expect(gated_store.uuids).to eq(%w[a]) + expect(b.batches_dropped?).to be(true) + end +end diff --git a/spec/unit/transcript_mirror_batcher_spec.rb b/spec/unit/transcript_mirror_batcher_spec.rb index 350bce5a..fb1e564e 100644 --- a/spec/unit/transcript_mirror_batcher_spec.rb +++ b/spec/unit/transcript_mirror_batcher_spec.rb @@ -165,8 +165,8 @@ def load(_key) = nil # Latency-injecting store: append sleeps on its (batcher-spawned) worker # thread, so a drain is in flight — holding the semaphore — while later # frames enqueue and schedule their own drains. This exercises the - # detach-before-lock + Async::Semaphore(1) ordering guarantee under genuine - # concurrency (the sole reason that machinery exists). + # detach-under-the-lock + Async::Semaphore(1) ordering guarantee under + # genuine concurrency (the sole reason that machinery exists). slow_store = Class.new(ClaudeAgentSDK::SessionStore) do def initialize super @@ -302,10 +302,11 @@ def load(_key) = @mutex.synchronize { @entries.dup } end # The looping drainer re-detaches after its append completes. A #flush - # barrier that detached OLDER frames is already parked on the lock by - # then, and Semaphore#release hands the lock straight to it (FIFO), so the - # drainer's NEWER frames must land after the barrier's — not ahead of it. - it 'appends a parked flush barrier batch before frames the drainer picks up later' do + # barrier is already parked on the lock by then; Semaphore#release hands + # the lock straight to it (FIFO) and it takes everything pending, older + # frames first — so nothing the drainer picks up later can land ahead of + # a frame that was enqueued before it. + it 'keeps enqueue order across a flush barrier parked behind the looping drainer' do Async do |task| task.with_timeout(10) do b = described_class.new(store: gated_store, projects_dir: projects, on_error: on_error, @@ -313,12 +314,12 @@ def load(_key) = @mutex.synchronize { @entries.dup } b.enqueue(file_path, [{ 'type' => 'user', 'uuid' => 'a' }]) gated_store.started.pop # drainer's append of `a` parked on the gate b.enqueue(file_path, [{ 'type' => 'user', 'uuid' => 'b' }]) - flusher = task.async { b.flush } # detaches `b`, parks on the lock - b.enqueue(file_path, [{ 'type' => 'user', 'uuid' => 'c' }]) # buffered for the drainer + flusher = task.async { b.flush } # parks on the lock + b.enqueue(file_path, [{ 'type' => 'user', 'uuid' => 'c' }]) # buffered behind `b` gated_store.gate.close flusher.wait - b.flush # barrier covering the drainer's `c` iteration + b.flush # barrier: nothing the drainer took is still in flight end end @@ -328,12 +329,14 @@ def load(_key) = @mutex.synchronize { @entries.dup } end # Teardown corner: Query#close runs batcher.close, then stops the read - # task (the drainer's parent). A frame the read loop enqueued during the - # close window sits in @pending (the live drainer would have taken it - # next), so stopping the read task first loses it — and that loss must - # surface through batches_dropped?, as the old per-frame parked drain's - # cancellation did, so resume-from-store teardown preserves the temp dir. - it 'counts a frame buffered during close as dropped when the read task is stopped first' do + # task (the drainer's parent). A frame the read loop enqueues while close + # is still queued behind the drainer's in-flight append used to sit in + # @pending until the read task was stopped, and was lost (reported through + # batches_dropped?). close detaches inside the lock now, so it takes that + # frame along and delivers it before the read task is stopped. (A frame + # that arrives after close detached is still a counted drop — see + # transcript_mirror_batcher_cancellation_spec.rb.) + it 'delivers a frame buffered while close waits behind an in-flight append, before the read task is stopped' do b = nil pending_at_stop = nil Async do |task| @@ -363,9 +366,9 @@ def load(_key) = @mutex.synchronize { @entries.dup } end end - expect(pending_at_stop).to eq(1) # the corner was reached: `late` never detached - expect(gated_store.entries.map { |e| e['uuid'] }).to eq(%w[a]) - expect(b.batches_dropped?).to be(true) + expect(pending_at_stop).to eq(0) # `late` went out with close's own batch + expect(gated_store.entries.map { |e| e['uuid'] }).to eq(%w[a late]) + expect(b.batches_dropped?).to be(false) end it 'does not report a drop when frames buffered during close are drained before teardown ends' do