diff --git a/internal/commands/agent_hook_connector.go b/internal/commands/agent_hook_connector.go index 841a0213d..4e5d2b0da 100644 --- a/internal/commands/agent_hook_connector.go +++ b/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 diff --git a/internal/commands/connect_run.go b/internal/commands/connect_run.go index 345de3318..c6aa0623c 100644 --- a/internal/commands/connect_run.go +++ b/internal/commands/connect_run.go @@ -347,6 +347,7 @@ 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}) @@ -354,7 +355,6 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error { 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, diff --git a/internal/connector/handoff.go b/internal/connector/handoff.go index 52abde89e..9bdbd5e45 100644 --- a/internal/connector/handoff.go +++ b/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) diff --git a/internal/connector/handoff_test.go b/internal/connector/handoff_test.go index 4ba37cd62..11c863313 100644 --- a/internal/connector/handoff_test.go +++ b/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) diff --git a/internal/connector/skill_layout_test.go b/internal/connector/skill_layout_test.go index 5257db680..338018943 100644 --- a/internal/connector/skill_layout_test.go +++ b/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") +} diff --git a/skills/basecamp-connect/SKILL.md b/skills/basecamp-connect/SKILL.md index 6206e9f01..dfaa3b971 100644 --- a/skills/basecamp-connect/SKILL.md +++ b/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 '' | grep --line-buffered -F '"type":"request"' +bash '${CLAUDE_SKILL_DIR}/watch-requests.sh' '' ~/.config/basecamp-connect/'handled-' ``` -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 >> ~/.config/basecamp-connect/'handled-'`), 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 ''` (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 ''`. -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 '' ~/.config/basecamp-connect/'handled-' +``` + +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 ''`: 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 ''`: 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. diff --git a/skills/basecamp-connect/watch-requests.sh b/skills/basecamp-connect/watch-requests.sh new file mode 100644 index 000000000..8656e77e0 --- /dev/null +++ b/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] " >&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