From f5341ed55785f762a721dc85fc489357b9337a9a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Wed, 2 Sep 2026 04:46:31 -0400 Subject: [PATCH] fix(go): retry rejected prev_chain_hash within a bounded budget and serialize chain-state updates; release 2.0.1 Claude-Session: https://claude.ai/code/session_01QoUjdc7HGKTsgj5GhG8HvF --- .gitignore | 1 + cmd/elydora/main.go | 2 +- cmd/elydora/plugins/claudecode_retry_test.go | 194 ++++++++++++++++ cmd/elydora/plugins/hooktemplate.go | 219 ++++++++++++------- 4 files changed, 340 insertions(+), 76 deletions(-) create mode 100644 cmd/elydora/plugins/claudecode_retry_test.go diff --git a/.gitignore b/.gitignore index 4c91543..65c6a74 100644 --- a/.gitignore +++ b/.gitignore @@ -31,3 +31,4 @@ Thumbs.db .env.local *.pem *.key +/elydora diff --git a/cmd/elydora/main.go b/cmd/elydora/main.go index e2b8e1b..d4552ec 100644 --- a/cmd/elydora/main.go +++ b/cmd/elydora/main.go @@ -11,7 +11,7 @@ import ( "github.com/Elydora-Infrastructure/Elydora-Go-SDK/v2/cmd/elydora/plugins" ) -const version = "2.0.0" +const version = "2.0.1" func main() { if len(os.Args) < 2 { diff --git a/cmd/elydora/plugins/claudecode_retry_test.go b/cmd/elydora/plugins/claudecode_retry_test.go new file mode 100644 index 0000000..f956eba --- /dev/null +++ b/cmd/elydora/plugins/claudecode_retry_test.go @@ -0,0 +1,194 @@ +package plugins + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +type mismatchAPI struct { + mu sync.Mutex + rejections int + delay time.Duration + staleHead bool + staleLock bool + liveLock bool + statePath string + operations []map[string]any +} + +const mismatchExpected = "Rxlf4j36C3KvIQ3hWuOkX698BR5iDypUFuB70JjEuvM" + +func (api *mismatchAPI) ServeHTTP(response http.ResponseWriter, request *http.Request) { + api.mu.Lock() + defer api.mu.Unlock() + response.Header().Set("Content-Type", "application/json") + if request.Method != http.MethodPost || request.URL.Path != "/v1/operations" { + _ = json.NewEncoder(response).Encode(map[string]any{"agent": map[string]any{"status": "active"}}) + return + } + var operation map[string]any + if err := json.NewDecoder(request.Body).Decode(&operation); err != nil { + http.Error(response, err.Error(), http.StatusBadRequest) + return + } + api.operations = append(api.operations, operation) + count := len(api.operations) + if api.delay > 0 { + api.mu.Unlock() + time.Sleep(api.delay) + api.mu.Lock() + } + if api.statePath != "" { + _ = os.WriteFile(api.statePath, []byte(`{"prev_chain_hash":"`+strings.Repeat("B", 43)+`"}`), 0o600) + } + if count > api.rejections { + response.WriteHeader(http.StatusAccepted) + _, _ = fmt.Fprintf(response, `{"receipt":{"seq_no":%d}}`, count) + return + } + expected := mismatchExpected + if api.rejections != 1 { + expected = strings.Repeat("A", 42) + fmt.Sprint(count) + } + response.WriteHeader(http.StatusBadRequest) + _, _ = fmt.Fprintf( + response, + `{"error":{"code":"PREV_HASH_MISMATCH","message":"Expected prev_chain_hash \"%s\", got \"x\"."}}`, + expected, + ) +} + +func runClaudeAuditAgainst(t *testing.T, api *mismatchAPI) (*claudeFixture, []map[string]any, time.Duration) { + t.Helper() + server := httptest.NewServer(api) + defer server.Close() + fixture := prepareClaudeFixture(t, claudeFixtureOptions{}) + fixture.config.BaseURL = server.URL + fixture.config.Token = "ely_test_token" + if err := fixture.plugin.Install(fixture.config); err != nil { + t.Fatalf("install Claude hooks: %v", err) + } + settings := readClaudeTestObject(t, fixture.configPath) + audit := claudeTestManagedHandler(t, settings, "PostToolUse", claudeAuditScript, claudeAuditStatusMessage) + if api.staleHead { + api.mu.Lock() + api.statePath = filepath.Join(fixture.agentDir, "chain-state.json") + api.mu.Unlock() + } + if api.staleLock || api.liveLock { + lockPath := filepath.Join(fixture.agentDir, "chain-state.json.lock") + var owner []byte + if api.liveLock { + owner = []byte(fmt.Sprint(os.Getpid())) + } + if err := os.WriteFile(lockPath, owner, 0o600); err != nil { + t.Fatalf("write stale lock: %v", err) + } + stale := time.Now().Add(-10 * time.Second) + if err := os.Chtimes(lockPath, stale, stale); err != nil { + t.Fatalf("age stale lock: %v", err) + } + } + started := time.Now() + exit, stdout, stderr := runClaudeHandler(t, audit, fixture, claudeOfficialPayload("PostToolUse")) + elapsed := time.Since(started) + if exit != 0 || stdout != "" || stderr != "" { + t.Fatalf("audit result = %d, %q, %q", exit, stdout, stderr) + } + api.mu.Lock() + defer api.mu.Unlock() + return fixture, api.operations, elapsed +} + +func TestClaudeAuditRetriesRejectedChainHash(t *testing.T) { + fixture, operations, _ := runClaudeAuditAgainst(t, &mismatchAPI{rejections: 1}) + if len(operations) != 2 || operations[1]["prev_chain_hash"] != mismatchExpected { + t.Fatalf("operations = %#v", operations) + } + if operations[0]["operation_id"] == operations[1]["operation_id"] || operations[0]["nonce"] == operations[1]["nonce"] { + t.Fatalf("retry reused identifiers: %#v", operations) + } + state, err := os.ReadFile(filepath.Join(fixture.agentDir, "chain-state.json")) + if err != nil { + t.Fatalf("read chain state: %v", err) + } + if !strings.Contains(string(state), fmt.Sprint(operations[1]["chain_hash"])) { + t.Fatalf("chain state %s does not hold %v", state, operations[1]["chain_hash"]) + } + log, err := os.ReadFile(filepath.Join(fixture.agentDir, "error.log")) + if err != nil || !strings.Contains(string(log), "resynced to server: Rxlf") { + t.Fatalf("error log = %q, %v", log, err) + } +} + +func TestClaudeAuditStopsAfterFiveRejections(t *testing.T) { + fixture, operations, _ := runClaudeAuditAgainst(t, &mismatchAPI{rejections: 99}) + if len(operations) != 5 || operations[4]["prev_chain_hash"] != strings.Repeat("A", 42)+"4" { + t.Fatalf("operations = %#v", operations) + } + log, err := os.ReadFile(filepath.Join(fixture.agentDir, "error.log")) + if err != nil || !strings.Contains(string(log), "rejected prev_chain_hash 5 times") { + t.Fatalf("error log = %q, %v", log, err) + } +} + +func TestClaudeAuditStopsRetryingWhenBudgetSpent(t *testing.T) { + fixture, operations, elapsed := runClaudeAuditAgainst(t, &mismatchAPI{rejections: 99, delay: 2600 * time.Millisecond}) + if elapsed > 7500*time.Millisecond { + t.Fatalf("hook ran %s", elapsed) + } + if len(operations) < 2 || len(operations) > 3 { + t.Fatalf("submissions = %d", len(operations)) + } + log, err := os.ReadFile(filepath.Join(fixture.agentDir, "error.log")) + if err != nil || !(strings.Contains(string(log), "aborted") || strings.Contains(string(log), "retry budget exhausted")) { + t.Fatalf("error log = %q, %v", log, err) + } +} + +func TestClaudeAuditDoesNotRegressAConcurrentlyAdvancedChainHead(t *testing.T) { + fixture, operations, _ := runClaudeAuditAgainst(t, &mismatchAPI{staleHead: true}) + if len(operations) != 1 || operations[0]["prev_chain_hash"] != strings.Repeat("A", 43) { + t.Fatalf("operations = %#v", operations) + } + state, err := os.ReadFile(filepath.Join(fixture.agentDir, "chain-state.json")) + if err != nil || !strings.Contains(string(state), strings.Repeat("B", 43)) { + t.Fatalf("chain state = %s, %v", state, err) + } +} + +func TestClaudeAuditClearsAStaleChainStateLock(t *testing.T) { + fixture, operations, _ := runClaudeAuditAgainst(t, &mismatchAPI{staleLock: true}) + if len(operations) != 1 { + t.Fatalf("operations = %#v", operations) + } + if _, err := os.Stat(filepath.Join(fixture.agentDir, "chain-state.json.lock")); !os.IsNotExist(err) { + t.Fatalf("stale lock still present: %v", err) + } + state, err := os.ReadFile(filepath.Join(fixture.agentDir, "chain-state.json")) + if err != nil || !strings.Contains(string(state), fmt.Sprint(operations[0]["chain_hash"])) { + t.Fatalf("chain state = %s, %v", state, err) + } +} + +func TestClaudeAuditDoesNotReclaimALiveOwnersLock(t *testing.T) { + fixture, operations, _ := runClaudeAuditAgainst(t, &mismatchAPI{liveLock: true}) + if len(operations) != 1 { + t.Fatalf("operations = %#v", operations) + } + if _, err := os.Stat(filepath.Join(fixture.agentDir, "chain-state.json.lock")); err != nil { + t.Fatalf("live lock was removed: %v", err) + } + log, err := os.ReadFile(filepath.Join(fixture.agentDir, "error.log")) + if err != nil || !strings.Contains(string(log), "lock timed out") { + t.Fatalf("error log = %q, %v", log, err) + } +} diff --git a/cmd/elydora/plugins/hooktemplate.go b/cmd/elydora/plugins/hooktemplate.go index 649e309..ff82f21 100644 --- a/cmd/elydora/plugins/hooktemplate.go +++ b/cmd/elydora/plugins/hooktemplate.go @@ -113,6 +113,8 @@ function signEd25519(privateKeyBase64url, data) { return base64urlEncode(signature); } +const MAX_CHAIN_ATTEMPTS = 5; +const SUBMIT_BUDGET_MS = 4000; const ZERO_CHAIN_HASH = 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; function validateChainHash(value, source) { @@ -188,6 +190,81 @@ function writeChainState(chainHash) { writeProtectedJson(CHAIN_STATE_PATH, 'Chain state', { prev_chain_hash: chainHash }); } +function lockOwnerAlive(lockPath) { + let owner; + try { + owner = Number.parseInt(fs.readFileSync(lockPath, 'utf-8'), 10); + } catch (error) { + if (!hasRuntimeErrorCode(error, 'ENOENT')) throw error; + return false; + } + if (!Number.isInteger(owner) || owner <= 0) return false; + try { + process.kill(owner, 0); + return true; + } catch (error) { + if (hasRuntimeErrorCode(error, 'ESRCH')) return false; + return true; + } +} + +function withChainLock(update) { + const lockPath = CHAIN_STATE_PATH + '.lock'; + const deadline = performance.now() + 1000; + for (;;) { + let descriptor; + try { + descriptor = fs.openSync(lockPath, 'wx', 0o600); + fs.writeSync(descriptor, String(process.pid)); + } catch (error) { + if (!hasRuntimeErrorCode(error, 'EEXIST')) throw error; + if (lockOwnerAlive(lockPath)) { + if (performance.now() > deadline) throw new Error('Chain state lock timed out: ' + lockPath); + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 10); + continue; + } + let age = 0; + try { age = Date.now() - fs.statSync(lockPath).mtimeMs; } catch (statError) { + if (!hasRuntimeErrorCode(statError, 'ENOENT')) throw statError; + } + if (age > 5000) { + const reclaimed = lockPath + '.' + process.pid + '.stale'; + try { + fs.renameSync(lockPath, reclaimed); + fs.unlinkSync(reclaimed); + } catch (reclaimError) { + if (!hasRuntimeErrorCode(reclaimError, 'ENOENT')) throw reclaimError; + } + continue; + } + if (performance.now() > deadline) throw new Error('Chain state lock timed out: ' + lockPath); + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 10); + continue; + } + try { + return update(); + } finally { + let current = null; + try { + current = fs.lstatSync(lockPath); + } catch (statError) { + if (!hasRuntimeErrorCode(statError, 'ENOENT')) throw statError; + } + const own = fs.fstatSync(descriptor); + fs.closeSync(descriptor); + if (current && current.ino === own.ino && current.dev === own.dev) fs.unlinkSync(lockPath); + } + } +} + +function advanceChainState(fromHash, toHash) { + return withChainLock(() => { + if (readChainState() !== fromHash) return false; + writeChainState(toHash); + return true; + }); +} + // --------------------------------------------------------------------------- // Error logging // --------------------------------------------------------------------------- @@ -230,6 +307,41 @@ function resolveAuditBaseURL(config) { // Main hook logic // --------------------------------------------------------------------------- +async function postOperation(baseUrl, token, eor, deadline) { + const remaining = deadline - performance.now(); + if (remaining <= 0) throw new Error('Audit API retry budget exhausted'); + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), Math.min(5000, remaining)); + try { + const headers = { 'Content-Type': 'application/json', 'Accept': 'application/json' }; + if (token) headers['Authorization'] = 'Bearer ' + token; + const res = await fetch(baseUrl + '/v1/operations', { + method: 'POST', + headers: headers, + body: JSON.stringify(eor), + signal: controller.signal, + }); + if (res.ok) return { accepted: true }; + if (res.status === 400) { + let errBody; + try { + errBody = await res.json(); + } catch (err) { + throw new Error('Audit API returned invalid error JSON: ' + err.message); + } + if (errBody.error && errBody.error.code === 'PREV_HASH_MISMATCH') { + const match = (errBody.error.message || '').match(/Expected prev_chain_hash "([^"]+)"/); + if (match) { + return { accepted: false, expected: validateChainHash(match[1], 'Audit API mismatch response') }; + } + } + } + throw new Error('Audit API returned HTTP ' + res.status); + } finally { + clearTimeout(timeout); + } +} + async function main() { try { // Read JSON from stdin @@ -304,91 +416,48 @@ async function main() { } if (typeof token !== 'string') throw new Error('Agent config token must be a string'); - // Read chain state - const prevChainHash = readChainState(); - - // Build EOR - const operationId = uuidv7(); - const issuedAt = Date.now(); - const nonce = generateNonce(); - + // Build and submit, retrying from the server chain head on PREV_HASH_MISMATCH const payload = NATIVE_PAYLOAD ? hookData : { tool_name: toolName, tool_input: toolInput, session_id: sessionId, }; - const payloadHash = computePayloadHash(payload); - const chainHash = computeChainHash(prevChainHash, payloadHash, operationId, issuedAt); - - const eorWithoutSig = { - op_version: '1.0', - operation_id: operationId, - org_id: orgId, - agent_id: agentId, - issued_at: issuedAt, - ttl_ms: 30000, - nonce: nonce, - operation_type: 'ai.tool_use', - subject: { session_id: sessionId }, - action: { tool: toolName }, - payload: payload, - payload_hash: payloadHash, - prev_chain_hash: prevChainHash, - agent_pubkey_kid: kid, - }; - - const canonical = jcsCanonicalise(eorWithoutSig); - const signature = signEd25519(privateKey, Buffer.from(canonical, 'utf-8')); - - const eor = Object.assign({}, eorWithoutSig, { chain_hash: chainHash, signature: signature }); - - // POST to /v1/operations (5s timeout) - // Only update local chain state on success to prevent desync - const url = baseUrl + '/v1/operations'; - const controller = new AbortController(); - const timeout = setTimeout(() => controller.abort(), 5000); - - try { - const headers = { - 'Content-Type': 'application/json', - 'Accept': 'application/json', + const deadline = performance.now() + SUBMIT_BUDGET_MS; + let prevChainHash = readChainState(); + for (let attempt = 1; ; attempt += 1) { + const operationId = uuidv7(); + const issuedAt = Date.now(); + const chainHash = computeChainHash(prevChainHash, payloadHash, operationId, issuedAt); + const eorWithoutSig = { + op_version: '1.0', + operation_id: operationId, + org_id: orgId, + agent_id: agentId, + issued_at: issuedAt, + ttl_ms: 30000, + nonce: generateNonce(), + operation_type: 'ai.tool_use', + subject: { session_id: sessionId }, + action: { tool: toolName }, + payload: payload, + payload_hash: payloadHash, + prev_chain_hash: prevChainHash, + agent_pubkey_kid: kid, }; - if (token) { - headers['Authorization'] = 'Bearer ' + token; - } - - const res = await fetch(url, { - method: 'POST', - headers: headers, - body: JSON.stringify(eor), - signal: controller.signal, - }); - - if (res.ok) { - writeChainState(chainHash); + const signature = signEd25519(privateKey, Buffer.from(jcsCanonicalise(eorWithoutSig), 'utf-8')); + const eor = Object.assign({}, eorWithoutSig, { chain_hash: chainHash, signature: signature }); + const result = await postOperation(baseUrl, token, eor, deadline); + if (result.accepted) { + advanceChainState(prevChainHash, chainHash); return; } - if (res.status === 400) { - // Possible chain hash mismatch — try to resync from server response - try { - const errBody = await res.json(); - if (errBody.error && errBody.error.code === 'PREV_HASH_MISMATCH') { - // Extract the expected hash from the error message - const match = (errBody.error.message || '').match(/Expected prev_chain_hash "([^"]+)"/); - if (match) { - const expectedHash = validateChainHash(match[1], 'Audit API mismatch response'); - writeChainState(expectedHash); - logError(new Error('Chain hash resynced to server: ' + expectedHash)); - } - } - } catch (err) { - logError(new Error('Failed to parse audit API error: ' + err.message)); - } + advanceChainState(prevChainHash, result.expected); + logError(new Error('Chain hash resynced to server: ' + result.expected)); + if (attempt >= MAX_CHAIN_ATTEMPTS) { + throw new Error('Audit API rejected prev_chain_hash ' + attempt + ' times'); } - throw new Error('Audit API returned HTTP ' + res.status); - } finally { - clearTimeout(timeout); + prevChainHash = readChainState(); } } catch (err) { logError(err);