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

internal/commands/agent_hook_connector.go

@@ -58,8 +58,9 @@ func connectorReloadNotice(input agentHookInput) string {
 		"finished cut that skill to its first 5,000 tokens, which drops the handoff steps and the subagent's " +
 		"instructions. If this session is still the connector's orchestrator, read " + skill + " in full with " +
 		"the Read tool now, before you go on with anything, a request in hand included, then carry on from the " +
-		"step you were on. The connector, its monitor and any subagents in flight kept running through the " +
-		"compaction: don't start them again."
+		"step you were on. The connector and any subagents in flight kept running through the compaction: don't " +
+		"start them again. The request watcher exits on each request, so check whether it is still running; if it " +
+		"has exited, read its output and start it again as the skill says."
 }
 
 // lastConnectorSkillDir returns the directory of the last basecamp-connect

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

@@ -75,25 +75,22 @@ see `connector: running` in its log. If it exited instead, explain the error
 (see *When the connector stops*), and stop. If it says `could not get the
 agent's token yet`, Basecamp's token endpoint is rate-limiting it or having
 trouble: it waits as long as it says and tries again on its own. Tell the
-person, leave it running, and arm the monitor as usual. Restarting it doesn't
+person, leave it running, and start the watcher as usual. Restarting it doesn't
 help.
 
-Then arm a persistent monitor on that file with the Monitor tool
-(`persistent: true`), replaying from the start so nothing that arrived between
-the start and the monitor is missed:
+Then start the watcher, also with `run_in_background: true`:
 
 ```bash
-tail -f -n +1 '<connector-output-file>' | grep --line-buffered -F '"type":"request"'
+bash '${CLAUDE_SKILL_DIR}/watch-requests.sh' '<connector-output-file>' ~/.config/basecamp-connect/'handled-<profile>'
 ```
 
-Single-quote the path, as with every other value.
-
-Each notification is one request: handle it with *For each request*. A request
-that arrives while you're waiting on the person is not their reply.
-
-The connector starts from the present. Anything asked more than a minute
-before it started is not picked up. If someone says the agent ignored them, that's the first
-thing to check.
+It exits once request lines arrive whose `event_id` the handled file lacks.
+Read them in full from its output file, record each
+(`echo <event_id> >> ~/.config/basecamp-connect/'handled-<profile>'`), start
+the watcher again, then handle each with *For each request*. Keep one running.
+Not the Monitor tool: its watches expire, and a `grep -v` in one holds a line
+back until the next. A request that arrives while you're waiting on the person
+is not their reply.
 
 ### 2. The request line
 
@@ -535,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),
@@ -547,15 +557,28 @@ 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:
+
+```bash
+bash '${CLAUDE_SKILL_DIR}/watch-requests.sh' --once '<old-output-file>' ~/.config/basecamp-connect/'handled-<profile>'
+```
+
+Handle any line it prints as usual. Start the watcher on the new file once the
+connector is running.
 
 ## Checking on it
 
 - `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?"**: 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
@@ -564,7 +587,8 @@ Requests that arrive while it's stopped are not picked up later.
 
 ## Stopping
 
-When the person asks you to stop, or the session is ending, stop 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.
+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 wait until it's started again, and
+are dropped if that's more than 4 hours later.

skills/basecamp-connect/watch-requests.sh

@@ -0,0 +1,50 @@
+#!/usr/bin/env bash
+set -u
+
+once=false
+if [ "${1:-}" = --once ]; then
+  once=true
+  shift
+fi
+if [ $# -ne 2 ]; then
+  echo "usage: watch-requests.sh [--once] <connector-output-file> <handled-file>" >&2
+  exit 2
+fi
+
+out=$1
+handled=$2
+
+if [ ! -f "$out" ]; then
+  echo "watch-requests: no connector output at $out" >&2
+  exit 2
+fi
+if ! { mkdir -p "$(dirname "$handled")" && touch "$handled" && [ -f "$handled" ] && [ -r "$handled" ] && [ -w "$handled" ]; }; then
+  echo "watch-requests: can't use the handled file $handled" >&2
+  exit 2
+fi
+
+unhandled() {
+  local lines
+  if [ ! -r "$out" ]; then
+    echo "watch-requests: can't read the connector output at $out" >&2
+    exit 2
+  fi
+  lines=$(($(wc -l < "$out")))
+  [ "$lines" -gt 0 ] || return 1
+  head -n "$lines" "$out" | HANDLED=$handled awk '
+    BEGIN { while ((getline id < ENVIRON["HANDLED"]) > 0) seen[id] = 1 }
+    /"type":"request"/ && match($0, /"event_id":[0-9]+/) {
+      id = substr($0, RSTART + 11, RLENGTH - 11)
+      if (!(id in seen)) { print; seen[id] = 1; found = 1 }
+    }
+    END { exit !found }'
+}
+
+if [ "$once" = true ]; then
+  unhandled
+  exit 0
+fi
+
+until unhandled; do
+  sleep 2
+done