Fork tasks from snapshots; snapshot only when text follows tool calls
This commit is contained in:
@@ -15,12 +15,15 @@ import (
|
||||
"g.fc.run/misc/byre/internal/store"
|
||||
)
|
||||
|
||||
const snapshotInterval = 30 * time.Second
|
||||
type snapRequest struct {
|
||||
trigger api.SnapshotTrigger
|
||||
seq int64
|
||||
}
|
||||
|
||||
type snapState struct {
|
||||
mu sync.Mutex
|
||||
running bool
|
||||
pending api.SnapshotTrigger
|
||||
pending *snapRequest
|
||||
closed bool
|
||||
}
|
||||
|
||||
@@ -33,7 +36,7 @@ func (m *Manager) snapStateLocked(taskID string) *snapState {
|
||||
return st
|
||||
}
|
||||
|
||||
func (m *Manager) snapshotSoon(l *live, trigger api.SnapshotTrigger) {
|
||||
func (m *Manager) snapshotSoon(l *live, trigger api.SnapshotTrigger, seq int64) {
|
||||
m.snapMu.Lock()
|
||||
defer m.snapMu.Unlock()
|
||||
|
||||
@@ -41,32 +44,33 @@ func (m *Manager) snapshotSoon(l *live, trigger api.SnapshotTrigger) {
|
||||
if st.closed {
|
||||
return
|
||||
}
|
||||
req := &snapRequest{trigger: trigger, seq: seq}
|
||||
if st.running {
|
||||
st.pending = trigger
|
||||
st.pending = req
|
||||
return
|
||||
}
|
||||
st.running = true
|
||||
go m.runSnapshots(l, st, trigger)
|
||||
go m.runSnapshots(l, st, req)
|
||||
}
|
||||
|
||||
func (m *Manager) runSnapshots(l *live, st *snapState, trigger api.SnapshotTrigger) {
|
||||
func (m *Manager) runSnapshots(l *live, st *snapState, req *snapRequest) {
|
||||
for {
|
||||
st.mu.Lock()
|
||||
m.snapMu.Lock()
|
||||
closed := st.closed
|
||||
m.snapMu.Unlock()
|
||||
if !closed {
|
||||
err := m.takeSnapshot(context.Background(), l, trigger)
|
||||
err := m.takeSnapshot(context.Background(), l, req.trigger, req.seq)
|
||||
if err != nil {
|
||||
slog.Error("take snapshot", "task", l.taskID, "trigger", trigger, "err", err)
|
||||
slog.Error("take snapshot", "task", l.taskID, "trigger", req.trigger, "err", err)
|
||||
}
|
||||
}
|
||||
st.mu.Unlock()
|
||||
|
||||
m.snapMu.Lock()
|
||||
trigger = st.pending
|
||||
st.pending = ""
|
||||
if trigger == "" || st.closed {
|
||||
req = st.pending
|
||||
st.pending = nil
|
||||
if req == nil || st.closed {
|
||||
st.running = false
|
||||
m.snapMu.Unlock()
|
||||
return
|
||||
@@ -75,10 +79,13 @@ func (m *Manager) runSnapshots(l *live, st *snapState, trigger api.SnapshotTrigg
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) takeSnapshot(ctx context.Context, l *live, trigger api.SnapshotTrigger) error {
|
||||
seq, err := m.store.LastTaskSeq(ctx, l.taskID)
|
||||
if err != nil {
|
||||
return err
|
||||
func (m *Manager) takeSnapshot(ctx context.Context, l *live, trigger api.SnapshotTrigger, seq int64) error {
|
||||
if seq == 0 {
|
||||
var err error
|
||||
seq, err = m.store.LastTaskSeq(ctx, l.taskID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
tree, err := git.Snapshot(l.dir)
|
||||
@@ -131,7 +138,6 @@ func (m *Manager) dropSnapshots(repo, taskID string) error {
|
||||
|
||||
func (m *Manager) afterTool(l *live) claude.HookFunc {
|
||||
return func(ctx context.Context, input json.RawMessage) (map[string]any, error) {
|
||||
m.snapshotSoon(l, api.SnapshotTriggerToolCall)
|
||||
text := m.deliverMidTurn(ctx, l)
|
||||
if text == "" {
|
||||
return map[string]any{}, nil
|
||||
@@ -144,24 +150,3 @@ func (m *Manager) afterTool(l *live) claude.HookFunc {
|
||||
}, nil
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) snapshotTimer(l *live) {
|
||||
ticker := time.NewTicker(snapshotInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-l.sess.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
sess, err := m.store.GetSession(context.Background(), l.id)
|
||||
if err != nil {
|
||||
slog.Error("check session for snapshot", "session", l.id, "err", err)
|
||||
continue
|
||||
}
|
||||
if sess.State == api.AgentSessionStateRunning {
|
||||
m.snapshotSoon(l, api.SnapshotTriggerTimer)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user