From 34bcc6392eecf7f03163d77e1849cdbe0c70af51 Mon Sep 17 00:00:00 2001 From: Yuichi Uemura Date: Sun, 12 Jul 2026 17:22:48 +0900 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Add=20claim/ack=20API=20+=20PID=20s?= =?UTF-8?q?entinel=20helper=20for=20atomic=20delivery?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Motivation External delivery daemons plug in as a receiver for grok-build's rule-file self-poll fallback or a future explicit-daemon mode. Today's inbox.sh --quiet reads unread rows and marks them read in two separate SQL statements. When such a daemon consumes messages, an inject failure between those two statements causes silent message loss, and a race between two receivers can cause duplicate delivery. This change adds a small claim/lease/ack API that lets any receiver atomically hold a message before consuming it: - inject failure -> release, next poll retries (no loss) - concurrent receivers -> only one wins the claim (no duplicate) - process crash -> lease expires, another receiver retries The API is orthogonal to the existing monitor watcher in grok-build: upstream's monitor watcher could adopt these helpers to atomicize its own delivery loop. Changes - scripts/internal/init-db.sh Add claims table (message_id PK, owner, expires_at) + expiry index. Uses CREATE TABLE IF NOT EXISTS so existing DBs upgrade in place. - scripts/lib/claims.sh (new, 52 lines) agmsg_claim_next / agmsg_ack_claim / agmsg_release_claim built around a single BEGIN IMMEDIATE transaction with INSERT OR IGNORE + expiry GC. - scripts/receiver-live.sh (new, 12 lines) Small helper: read pid=N from an agent's ready sentinel and kill -0 it. Exits 0 if live, non-zero if stale or missing. Rule files or watchers can use it to gate a fallback path. - tests/test_claims.bats (new, 5 cases) Excludes already-claimed / release + reclaim + ack / wrong-owner ack rejection / expired claim reclaim / newline+tab body round-trip. Notes - No existing behavior is changed; nothing calls the new API yet. - Driver-side integration is planned as a follow-up so it can be discussed against the current monitor design. --- scripts/internal/init-db.sh | 7 +++++ scripts/lib/claims.sh | 52 ++++++++++++++++++++++++++++++++ scripts/receiver-live.sh | 12 ++++++++ tests/test_claims.bats | 59 +++++++++++++++++++++++++++++++++++++ 4 files changed, 130 insertions(+) create mode 100644 scripts/lib/claims.sh create mode 100755 scripts/receiver-live.sh create mode 100644 tests/test_claims.bats diff --git a/scripts/internal/init-db.sh b/scripts/internal/init-db.sh index 940a63a3..c6c74566 100755 --- a/scripts/internal/init-db.sh +++ b/scripts/internal/init-db.sh @@ -35,4 +35,11 @@ CREATE TABLE IF NOT EXISTS messages ( CREATE INDEX IF NOT EXISTS idx_unread ON messages(team, to_agent, read_at) WHERE read_at IS NULL; CREATE INDEX IF NOT EXISTS idx_history ON messages(team, created_at DESC); + +CREATE TABLE IF NOT EXISTS claims ( + message_id INTEGER PRIMARY KEY, + owner TEXT NOT NULL, + expires_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_claims_expiry ON claims(expires_at); SQL diff --git a/scripts/lib/claims.sh b/scripts/lib/claims.sh new file mode 100644 index 00000000..fafa9616 --- /dev/null +++ b/scripts/lib/claims.sh @@ -0,0 +1,52 @@ +#!/usr/bin/env bash +# delivery daemon 向け unread message claim API。 + +if ! type agmsg_db_path >/dev/null 2>&1; then + _agmsg_claims_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + source "$_agmsg_claims_dir/storage.sh" +fi + +_agmsg_claim_quote() { printf "'%s'" "$(printf '%s' "$1" | sed "s/'/''/g")"; } + +_agmsg_claim_init() { + local scripts_dir + scripts_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" + bash "$scripts_dir/internal/init-db.sh" >/dev/null +} + +# stdout: id US from US body US created_at。空 output は claim 対象なし。 +agmsg_claim_next() { + local team="$1" agent="$2" owner="$3" ttl="${4:-30}" db + db="$(agmsg_db_path)"; _agmsg_claim_init + agmsg_sqlite "$db" </dev/null diff --git a/tests/test_claims.bats b/tests/test_claims.bats new file mode 100644 index 00000000..5890a85a --- /dev/null +++ b/tests/test_claims.bats @@ -0,0 +1,59 @@ +#!/usr/bin/env bats + +load test_helper + +setup() { + setup_test_env + source "$SCRIPTS/lib/claims.sh" + export TEST_TEAM=claims-team TEST_AGENT=grok-pane TEST_OWNER=daemon-a + "$SCRIPTS/send.sh" "$TEST_TEAM" sender "$TEST_AGENT" "claim payload" >/dev/null +} + +teardown() { teardown_test_env; } + +@test "claim is exclusive until release" { + run agmsg_claim_next "$TEST_TEAM" "$TEST_AGENT" "$TEST_OWNER" 60 + [ "$status" -eq 0 ] + [ -n "$output" ] + + run agmsg_claim_next "$TEST_TEAM" "$TEST_AGENT" daemon-b 60 + [ "$status" -eq 0 ] + [ -z "$output" ] +} + +@test "released claim can be claimed and acknowledged exactly once" { + first="$(agmsg_claim_next "$TEST_TEAM" "$TEST_AGENT" "$TEST_OWNER" 60)" + id="${first%%$'\x1f'*}" + agmsg_release_claim "$id" "$TEST_OWNER" + + second="$(agmsg_claim_next "$TEST_TEAM" "$TEST_AGENT" daemon-b 60)" + [ -n "$second" ] + id="${second%%$'\x1f'*}" + agmsg_ack_claim "$id" daemon-b + + run agmsg_claim_next "$TEST_TEAM" "$TEST_AGENT" daemon-c 60 + [ "$status" -eq 0 ] + [ -z "$output" ] +} + +@test "wrong owner cannot acknowledge a claim" { + first="$(agmsg_claim_next "$TEST_TEAM" "$TEST_AGENT" "$TEST_OWNER" 60)" + id="${first%%$'\x1f'*}" + agmsg_ack_claim "$id" wrong-owner + run agmsg_claim_next "$TEST_TEAM" "$TEST_AGENT" daemon-b 60 + [ -z "$output" ] +} + +@test "expired claim is reclaimed by another daemon" { + first="$(agmsg_claim_next "$TEST_TEAM" "$TEST_AGENT" "$TEST_OWNER" 0)" + [ -n "$first" ] + run agmsg_claim_next "$TEST_TEAM" "$TEST_AGENT" daemon-b 60 + [ "$status" -eq 0 ] + [ -n "$output" ] +} + +@test "claim preserves escaped body text" { + "$SCRIPTS/send.sh" "$TEST_TEAM" sender "$TEST_AGENT" $'line one\nline two\tend' >/dev/null + first="$(agmsg_claim_next "$TEST_TEAM" "$TEST_AGENT" "$TEST_OWNER" 60)" + [[ "$first" == *'line one\\nline two\\tend'* ]] +}