Class: RobotLab::To::Guards::Checkpoint

Inherits:
Hook
  • Object
show all
Defined in:
lib/robot_lab/to/guards/checkpoint.rb

Overview

First-write-wins file snapshot before Write/Edit, ported from little-coder checkpoint. Finer-grained than robot_lab-to's git rollback: it survives WITHIN an iteration, so the model (or a future repair step) can revert a single botched edit without discarding the iteration's good work. Best-effort — never raises into the tool path.

Backups land under <run_dir>/checkpoints/, falling back to a temp dir when no Run is supplied. Files that did not exist get a .absent sentinel. The snapshotted-set spans calls, so it lives in RunStore.

Constant Summary collapse

KEY = 
:robot_lab_to_checkpoint
MUTATING_TOOLS = 
%w[write edit].freeze

Class Method Summary collapse

Class Method Details

.before_run(_ctx) ⇒ Object 



27
28
29
 
# File 'lib/robot_lab/to/guards/checkpoint.rb', line 27

def before_run(_ctx)
  RunStore.reset(KEY, [])
end

.before_tool_call(ctx) ⇒ Object 



31
32
33
34
35
36
37
38
 
# File 'lib/robot_lab/to/guards/checkpoint.rb', line 31

def before_tool_call(ctx)
  return unless MUTATING_TOOLS.include?(ctx.tool_name.to_s.downcase)

  path = PathResolution.resolve(ctx.tool_args)
  return unless path

  snapshot(path, checkpoint_dir(ctx))
end

.checkpoint_dir(ctx) ⇒ String

Returns directory for this run's checkpoints (created).

Returns:

  • (String) —

    directory for this run's checkpoints (created)



64
65
66
67
68
69
70
 
# File 'lib/robot_lab/to/guards/checkpoint.rb', line 64

def checkpoint_dir(ctx)
  run  = ctx.local.run
  base = run.respond_to?(:run_dir) ? run.run_dir : Dir.tmpdir
  dir  = File.join(base.to_s, "checkpoints")
  FileUtils.mkdir_p(dir)
  dir
end

.safe_name(path) ⇒ String

Flatten an absolute path into a single safe filename.

Parameters:

  • path (String)

Returns:

  • (String) 


76
77
78
79
 
# File 'lib/robot_lab/to/guards/checkpoint.rb', line 76

def safe_name(path)
  flat = path.gsub(/[^A-Za-z0-9._-]/, "_")
  flat.length > 200 ? flat[-200..] : flat
end

.snapshot(path, dir) ⇒ String?

Copy path's current bytes into dir, once per run (first-write wins). Swallows all I/O errors — checkpointing must never break a tool call.

Returns:

  • (String, nil) —

    the backup path written, or nil if skipped



45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/robot_lab/to/guards/checkpoint.rb', line 45

def snapshot(path, dir)
  tracked = RunStore.fetch(KEY, [])
  return if tracked.include?(path)

  tracked << path
  FileUtils.mkdir_p(dir)
  dest = File.join(dir, safe_name(path))
  if File.exist?(path)
    FileUtils.cp(path, dest)
    dest
  else
    File.write("#{dest}.absent", "")
    "#{dest}.absent"
  end
rescue SystemCallError, IOError
  nil
end