Catch up on mentions made while the connector was down, and restart it after a memory stop

internal/commands/connect_run.go

@@ -347,14 +347,14 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error {
 	if err := ledger.NoteConnection(ctx, connector.ConnectionRunning, ""); err != nil {
 		logger.Warn("connector: could not record that it runs, for status", "error", err)
 	}
+	started := time.Now()
 	runPart("intake", intake.Run)
 	runPart("admission", func(ctx context.Context) error {
 		return connector.RunAdmission(ctx, connector.AdmissionOptions{Ledger: ledger, Queue: queue, Admitter: admitter, Lines: lines, Logger: logger})
 	})
 	if !f.shadow {
 		// Each trusted request goes to stdout for the session that started
 		// the connector, and the connector is done with it.
-		started := time.Now()
 		runPart("handoff", func(ctx context.Context) error {
 			return connector.RunHandoff(ctx, connector.HandoffOptions{
 				Ledger: ledger, Served: served.Current, Buckets: buckets, AgentID: agentID,

internal/connector/handoff.go

@@ -76,8 +76,8 @@ const (
 	// ReasonUnreadable: the request's record had no content to hand off, so
 	// nothing was written.
 	ReasonUnreadable = "unreadable"
-	// ReasonBeforeThisRun: the request was made before this run started.
-	// Nobody was there to take it, and it is not run late.
+	// ReasonBeforeThisRun: the request was made too long before this run
+	// started to answer now.
 	ReasonBeforeThisRun = "before_this_run"
 )
 
@@ -86,6 +86,8 @@ const (
 // session started is still meant for it.
 const HandoffGrace = time.Minute
 
+const handoffCatchUp = 4 * time.Hour
+
 // HandoffOptions configure RunHandoff.
 type HandoffOptions struct {
 	Ledger *Ledger
@@ -178,11 +180,18 @@ func handOffReady(ctx context.Context, opts HandoffOptions) error {
 	}
 }
 
-// handOff writes one record and closes it. A record made before the run
-// started is closed without being written.
+func handoffCutoff(opts HandoffOptions, record Record) time.Time {
+	if decided := record.Decision.DecidedAt; decided != nil && !decided.Before(opts.Started) {
+		return opts.Started.Add(-handoffCatchUp)
+	}
+	return opts.Started.Add(-HandoffGrace)
+}
+
+// handOff writes one record and closes it. A record made too long before the
+// run started is closed without being written.
 func handOff(ctx context.Context, opts HandoffOptions, record Record) error {
-	if record.CreatedAt.Before(opts.Started.Add(-HandoffGrace)) {
-		opts.log().Info("connector: a request from before this run started is not handed off", "event_id", record.ID, "created_at", record.CreatedAt)
+	if cutoff := handoffCutoff(opts, record); record.CreatedAt.Before(cutoff) {
+		opts.log().Info("connector: a request made too long before this run started is not handed off", "event_id", record.ID, "created_at", record.CreatedAt, "cutoff", cutoff)
 		return opts.Ledger.SetState(ctx, record.ID, StateDiscarded, ReasonBeforeThisRun)
 	}
 	line, err := handoffLine(record, opts.AgentID)

internal/connector/handoff_test.go

@@ -76,14 +76,37 @@ func TestEachTrustedRequestIsHandedOffOnce(t *testing.T) {
 	}
 }
 
-// A request made before this run started is not run late: nobody was there
-// to take it.
-func TestARequestFromBeforeTheRunIsNotHandedOff(t *testing.T) {
+func admitAt(t *testing.T, ledger *Ledger, id int64, at time.Time) {
+	t.Helper()
+	ledger.now = func() time.Time { return at }
+	admitOn(t, ledger, id, "recording:1")
+}
+
+func TestARequestThisRunAdmittedIsHandedOffIfMadeWithinTheCatchUp(t *testing.T) {
+	for _, before := range []time.Duration{2 * time.Minute, time.Hour, handoffCatchUp} {
+		t.Run(before.String(), func(t *testing.T) {
+			ledger := newTestLedger(t)
+			var out bytes.Buffer
+			opts := handoffOptions(ledger, &out)
+			opts.Started = testEvent(1).CreatedAt.Add(before)
+			admitAt(t, ledger, 1, opts.Started.Add(time.Second))
+
+			require.NoError(t, handOffReady(context.Background(), opts))
+
+			lines := handedOffLines(t, &out)
+			require.Len(t, lines, 1)
+			assert.Equal(t, int64(1), lines[0].EventID)
+			assert.Equal(t, ReasonHandedOff, getRecord(t, ledger, 1).Reason)
+		})
+	}
+}
+
+func TestARequestThisRunAdmittedIsClosedIfMadeBeforeTheCatchUp(t *testing.T) {
 	ledger := newTestLedger(t)
-	admitOn(t, ledger, 1, "recording:1")
 	var out bytes.Buffer
 	opts := handoffOptions(ledger, &out)
-	opts.Started = opts.Started.Add(time.Hour)
+	opts.Started = testEvent(1).CreatedAt.Add(handoffCatchUp + time.Second)
+	admitAt(t, ledger, 1, opts.Started.Add(time.Second))
 
 	require.NoError(t, handOffReady(context.Background(), opts))
 
@@ -93,6 +116,37 @@ func TestARequestFromBeforeTheRunIsNotHandedOff(t *testing.T) {
 	assert.Equal(t, ReasonBeforeThisRun, r.Reason)
 }
 
+func TestARequestAnEarlierRunAdmittedIsHandedOffOnlyWithinTheGrace(t *testing.T) {
+	for _, tc := range []struct {
+		before time.Duration
+		handed bool
+	}{
+		{HandoffGrace, true},
+		{HandoffGrace + time.Second, false},
+		{time.Hour, false},
+	} {
+		t.Run(tc.before.String(), func(t *testing.T) {
+			ledger := newTestLedger(t)
+			var out bytes.Buffer
+			opts := handoffOptions(ledger, &out)
+			opts.Started = testEvent(1).CreatedAt.Add(tc.before)
+			admitAt(t, ledger, 1, testEvent(1).CreatedAt.Add(time.Second))
+
+			require.NoError(t, handOffReady(context.Background(), opts))
+
+			r := getRecord(t, ledger, 1)
+			if tc.handed {
+				require.Len(t, handedOffLines(t, &out), 1)
+				assert.Equal(t, ReasonHandedOff, r.Reason)
+				return
+			}
+			assert.Empty(t, out.String())
+			assert.Equal(t, StateDiscarded, r.State)
+			assert.Equal(t, ReasonBeforeThisRun, r.Reason)
+		})
+	}
+}
+
 // Only the projects the agent serves, narrowed to the run's --project.
 func TestOnlyServedProjectsAreHandedOff(t *testing.T) {
 	ledger := newTestLedger(t)

internal/connector/skill_layout_test.go

@@ -1,10 +1,12 @@
 package connector
 
 import (
+	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -93,3 +95,36 @@ func TestTheSubagentReadsItsInstructionsFromTheirOwnFile(t *testing.T) {
 		assert.Contains(t, subagent, step)
 	}
 }
+
+func flatText(s string) string {
+	return strings.Join(strings.Fields(s), " ")
+}
+
+func TestTheConnectSkillStartsTheConnectorAgainAfterAMemoryStop(t *testing.T) {
+	skill := readConnectSkill(t, "SKILL.md")
+	start := strings.Index(skill, "## When the connector stops")
+	require.GreaterOrEqual(t, start, 0)
+	end := strings.Index(skill[start+1:], "\n## ")
+	require.GreaterOrEqual(t, end, 0)
+	section := flatText(skill[start : start+1+end])
+
+	assert.Contains(t, section, "stopped because the system is running low on memory")
+	assert.Contains(t, section, "So start it again at once, as in step 1")
+	assert.Contains(t, section, "When such a notice names the watcher, check the connector first. If it was stopped too, start it again as above, which starts a new watcher, and don't start one on the old file.")
+	assert.Contains(t, section, "If it's still running, start the watcher again on its current output file.")
+}
+
+func TestTheConnectSkillStatesHowFarBackARestartCatchesUp(t *testing.T) {
+	skill := flatText(readConnectSkill(t, "SKILL.md"))
+	window := fmt.Sprintf("%d hours", int(handoffCatchUp/time.Hour))
+
+	for _, phrase := range []string{
+		"it hands off the requests made while it was stopped, up to " + window + " before it started.",
+		"hands off requests made up to " + window + " before it started",
+		"dropped if that's more than " + window + " later.",
+	} {
+		assert.Contains(t, skill, phrase)
+	}
+	assert.NotContains(t, skill, "not picked up later")
+	assert.NotContains(t, skill, "starts from the present")
+}

skills/basecamp-connect/SKILL.md

@@ -532,8 +532,21 @@ first. Every change goes through setup, which keeps whatever you don't pass.
 
 ## When the connector stops
 
-The background task notifies you when the connector exits. Read the end of its
-output and tell the person what it means:
+The background task notifies you when the connector exits.
+
+If the notice says it was **stopped because the system is running low on
+memory**, Claude Code stopped it while the session sat idle. The connector
+didn't fail. The notice says not to start it again on your own, but the person
+asked you to keep the connector running when they started it. So start it again
+at once, as in step 1, check the old output file and start the watcher on the
+new one as the end of this section says, and tell the person it happened. A
+command started again soon after such a stop isn't stopped again until someone
+types in the session. When such a notice names the watcher, check the
+connector first. If it was stopped too, start it again as above, which starts a
+new watcher, and don't start one on the old file. If it's still running, start
+the watcher again on its current output file.
+
+Otherwise, read the end of its output and tell the person what it means:
 
 - **"was disconnected in Basecamp, or connected on another computer"**: run
   `basecamp connect setup -P '<profile>'` (it reconnects with one approval),
@@ -544,7 +557,8 @@ output and tell the person what it means:
 - **Anything else**: show the error, and run `basecamp connect doctor -P
   '<profile>'`.
 
-Requests that arrive while it's stopped are not picked up later.
+When it starts again, it hands off the requests made while it was stopped, up
+to 4 hours before it started. Older ones are closed unanswered.
 
 Each start writes a new output file. Stop the watcher (TaskStop), then check
 the old file once, so a request printed just before the exit isn't left there:
@@ -561,10 +575,10 @@ connector is running.
 - `basecamp connect doctor -P '<profile>'`: the credential, the identity, the
   event feed and the ledger. Start here when something seems wrong. It posts
   nothing.
-- **"Did that mention reach it?"**: the connector starts from the present, so
-  anything asked more than a minute before it started is not picked up. Check
-  that first. Then search the connector's output file for the
-  recording id. A `"type":"request"` line means it was handed to you. A
+- **"Did that mention reach it?"**: the connector hands off requests made up
+  to 4 hours before it started, so anything asked earlier, while it was
+  stopped, is not picked up. Check that first. Then search the connector's
+  output file for the recording id. A `"type":"request"` line means it was handed to you. A
   `"type":"event"` line with a `reason` says why it was turned away (for
   example `untrusted_performer`). No line means the feed never delivered it.
 - `basecamp connect status -P '<profile>'`: the feed's position, holds and
@@ -576,5 +590,5 @@ connector is running.
 When the person asks you to stop, or the session is ending, stop the watcher
 and the connector task (TaskStop, or send it SIGINT). There's nothing else to
 clean up: no webhooks, nothing public. Subagents already running finish their
-requests and reply. Say that new mentions won't be picked up until it's
-started again.
+requests and reply. Say that new mentions wait until it's started again, and
+are dropped if that's more than 4 hours later.