fix(decision): export receipts appended by hook processes

decision/receipt_export.go

@@ -121,6 +121,9 @@ func (exporter *ReceiptExporter) ExportOnce(ctx context.Context) error {
 	}
 	exporter.mu.Lock()
 	defer exporter.mu.Unlock()
+	if err := exporter.journal.Refresh(); err != nil {
+		return fmt.Errorf("decision: refresh receipt journal: %w", err)
+	}
 	attempted := 0
 	var firstErr error
 	for _, receipt := range exporter.journal.Receipts() {

decision/receipt_export_test.go

@@ -7,11 +7,72 @@ import (
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
+	"os"
 	"path/filepath"
+	"strings"
 	"sync/atomic"
 	"testing"
 )
 
+func TestReceiptExporterDiscoversEvidenceAppendedByAnotherProcess(t *testing.T) {
+	t.Parallel()
+	path := filepath.Join(t.TempDir(), "receipts.jsonl")
+	exporterJournal, err := OpenReceiptJournal(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	writerJournal, err := OpenReceiptJournal(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	receipt := journalReceipt(t, 1785500000, Enforced)
+	if err := writerJournal.AppendReceipt(context.Background(), receipt); err != nil {
+		t.Fatal(err)
+	}
+	var calls atomic.Int64
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		calls.Add(1)
+		if request.Header.Get("Idempotency-Key") != receipt.ID {
+			t.Errorf("idempotency key=%q", request.Header.Get("Idempotency-Key"))
+		}
+		_ = json.NewEncoder(writer).Encode(map[string]string{"accepted_receipt_id": receipt.ID})
+	}))
+	defer server.Close()
+	exporter, err := NewReceiptExporter(ReceiptExporterConfig{
+		Journal: exporterJournal, Endpoint: server.URL, AckPath: filepath.Join(t.TempDir(), "acks"),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := exporter.ExportOnce(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if calls.Load() != 1 || exporter.Pending() != 0 {
+		t.Fatalf("external receipt calls=%d pending=%d", calls.Load(), exporter.Pending())
+	}
+}
+
+func TestReceiptExporterSurfacesIncompleteExternalJournalRecord(t *testing.T) {
+	t.Parallel()
+	path := filepath.Join(t.TempDir(), "receipts.jsonl")
+	journal, err := OpenReceiptJournal(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	exporter, err := NewReceiptExporter(ReceiptExporterConfig{
+		Journal: journal, Endpoint: "http://127.0.0.1:1/receipts", AckPath: filepath.Join(t.TempDir(), "acks"),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(path, []byte("{"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if err := exporter.ExportOnce(context.Background()); err == nil || !strings.Contains(err.Error(), "refresh receipt journal") {
+		t.Fatalf("incomplete external journal error=%v", err)
+	}
+}
+
 func TestReceiptExporterRetriesAndAcknowledgesSignedEvidence(t *testing.T) {
 	t.Parallel()
 	journal, err := OpenReceiptJournal(filepath.Join(t.TempDir(), "receipts.jsonl"))

decision/receipt_journal.go

@@ -27,6 +27,7 @@ type ReceiptJournal struct {
 	mu       sync.Mutex
 	seen     map[string]string
 	receipts []Receipt
+	offset   int64
 }
 
 func OpenReceiptJournal(path string) (*ReceiptJournal, error) {
@@ -46,7 +47,7 @@ func OpenReceiptJournal(path string) (*ReceiptJournal, error) {
 		return nil, fmt.Errorf("decision: create receipt journal directory: %w", err)
 	}
 	journal := &ReceiptJournal{path: absolute, seen: make(map[string]string)}
-	if err := journal.load(); err != nil {
+	if err := journal.refreshLocked(); err != nil {
 		return nil, err
 	}
 	return journal, nil
@@ -107,11 +108,31 @@ func (journal *ReceiptJournal) Receipts() []Receipt {
 	return append([]Receipt(nil), journal.receipts...)
 }
 
-func (journal *ReceiptJournal) load() error {
-	file, err := os.Open(journal.path)
+// Refresh discovers complete records appended by another process after this
+// journal was opened. Managed hooks and the long-running daemon deliberately
+// use separate processes, so exporters must refresh before inspecting their
+// in-memory snapshot. Existing records are not rescanned on every poll.
+func (journal *ReceiptJournal) Refresh() error {
+	if journal == nil || journal.path == "" {
+		return fmt.Errorf("decision: receipt journal is not initialized")
+	}
+	journal.mu.Lock()
+	defer journal.mu.Unlock()
+	return journal.refreshLocked()
+}
+
+func (journal *ReceiptJournal) refreshLocked() error {
+	pathInfo, err := os.Lstat(journal.path)
 	if os.IsNotExist(err) {
 		return nil
 	}
+	if err != nil {
+		return fmt.Errorf("decision: inspect receipt journal: %w", err)
+	}
+	if pathInfo.Mode()&os.ModeSymlink != 0 {
+		return fmt.Errorf("decision: receipt journal must not be a symlink")
+	}
+	file, err := os.Open(journal.path)
 	if err != nil {
 		return fmt.Errorf("decision: open receipt journal: %w", err)
 	}
@@ -123,9 +144,28 @@ func (journal *ReceiptJournal) load() error {
 	if info.Mode().Perm()&0o077 != 0 {
 		return fmt.Errorf("decision: receipt journal permissions must be owner-only")
 	}
-	scanner := bufio.NewScanner(file)
+	if info.Size() < journal.offset {
+		return fmt.Errorf("decision: receipt journal was truncated")
+	}
+	if info.Size() == journal.offset {
+		return nil
+	}
+	var last [1]byte
+	if _, err := file.ReadAt(last[:], info.Size()-1); err != nil {
+		return fmt.Errorf("decision: inspect receipt journal boundary: %w", err)
+	}
+	if last[0] != '\n' {
+		return fmt.Errorf("decision: receipt journal has an incomplete trailing record")
+	}
+	scanner := bufio.NewScanner(io.NewSectionReader(file, journal.offset, info.Size()-journal.offset))
 	scanner.Buffer(make([]byte, 64<<10), MaxReceiptJournalLineBytes+1)
-	line := 0
+	line := len(journal.receipts)
+	type pendingReceipt struct {
+		receipt Receipt
+		hash    string
+	}
+	pending := make([]pendingReceipt, 0)
+	pendingSeen := make(map[string]string)
 	for scanner.Scan() {
 		line++
 		decoder := json.NewDecoder(bytes.NewReader(scanner.Bytes()))
@@ -148,12 +188,26 @@ func (journal *ReceiptJournal) load() error {
 		if existing, exists := journal.seen[receipt.ID]; exists && existing != hash {
 			return fmt.Errorf("decision: conflicting receipt journal id %q", receipt.ID)
 		}
-		journal.seen[receipt.ID] = hash
-		journal.receipts = append(journal.receipts, receipt)
+		if existing, exists := pendingSeen[receipt.ID]; exists && existing != hash {
+			return fmt.Errorf("decision: conflicting receipt journal id %q", receipt.ID)
+		}
+		if existing, exists := journal.seen[receipt.ID]; exists && existing == hash {
+			continue
+		}
+		if existing, exists := pendingSeen[receipt.ID]; exists && existing == hash {
+			continue
+		}
+		pendingSeen[receipt.ID] = hash
+		pending = append(pending, pendingReceipt{receipt: receipt, hash: hash})
 	}
 	if err := scanner.Err(); err != nil {
 		return fmt.Errorf("decision: scan receipt journal: %w", err)
 	}
+	for _, record := range pending {
+		journal.seen[record.receipt.ID] = record.hash
+		journal.receipts = append(journal.receipts, record.receipt)
+	}
+	journal.offset = info.Size()
 	return nil
 }
 

decision/receipt_journal_test.go

@@ -6,10 +6,14 @@ import (
 	"context"
 	"crypto/ed25519"
 	"crypto/rand"
+	"encoding/json"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
+
+	"github.com/pilot-protocol/common/fsutil"
 )
 
 func journalReceipt(t *testing.T, observedAt int64, result EnforcementResult) Receipt {
@@ -107,3 +111,92 @@ func TestReceiptJournalRejectsUnsignedCorruptAndUnsafeFiles(t *testing.T) {
 		t.Fatal("symlink journal was accepted")
 	}
 }
+
+func TestReceiptJournalRefreshRejectsUnsafeExternalChanges(t *testing.T) {
+	t.Parallel()
+	if err := (*ReceiptJournal)(nil).Refresh(); err == nil {
+		t.Fatal("nil journal refresh succeeded")
+	}
+	t.Run("incomplete record", func(t *testing.T) {
+		path := filepath.Join(t.TempDir(), "receipts.jsonl")
+		journal, err := OpenReceiptJournal(path)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := os.WriteFile(path, []byte("{"), 0o600); err != nil {
+			t.Fatal(err)
+		}
+		if err := journal.Refresh(); err == nil || !strings.Contains(err.Error(), "incomplete trailing record") {
+			t.Fatalf("incomplete refresh error=%v", err)
+		}
+	})
+	t.Run("truncated journal", func(t *testing.T) {
+		path := filepath.Join(t.TempDir(), "receipts.jsonl")
+		journal, err := OpenReceiptJournal(path)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := journal.AppendReceipt(context.Background(), journalReceipt(t, 1785500000, Enforced)); err != nil {
+			t.Fatal(err)
+		}
+		if err := journal.Refresh(); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.Truncate(path, 0); err != nil {
+			t.Fatal(err)
+		}
+		if err := journal.Refresh(); err == nil || !strings.Contains(err.Error(), "truncated") {
+			t.Fatalf("truncated refresh error=%v", err)
+		}
+	})
+	t.Run("symlink replacement", func(t *testing.T) {
+		directory := t.TempDir()
+		path := filepath.Join(directory, "receipts.jsonl")
+		journal, err := OpenReceiptJournal(path)
+		if err != nil {
+			t.Fatal(err)
+		}
+		target := filepath.Join(directory, "target.jsonl")
+		if err := os.WriteFile(target, nil, 0o600); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.Symlink(target, path); err != nil {
+			t.Fatal(err)
+		}
+		if err := journal.Refresh(); err == nil || !strings.Contains(err.Error(), "symlink") {
+			t.Fatalf("symlink refresh error=%v", err)
+		}
+	})
+}
+
+func TestReceiptJournalRefreshRejectsConflictingExternalRecord(t *testing.T) {
+	t.Parallel()
+	path := filepath.Join(t.TempDir(), "receipts.jsonl")
+	journal, err := OpenReceiptJournal(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	receipt := journalReceipt(t, 1785500000, Enforced)
+	if err := journal.AppendReceipt(context.Background(), receipt); err != nil {
+		t.Fatal(err)
+	}
+	if err := journal.Refresh(); err != nil {
+		t.Fatal(err)
+	}
+	conflict := receipt
+	conflict.ObservedAt++
+	_, privateKey, _ := ed25519.GenerateKey(rand.Reader)
+	if err := conflict.Sign(privateKey); err != nil {
+		t.Fatal(err)
+	}
+	body, err := json.Marshal(conflict)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := fsutil.AppendSync(path, append(body, '\n')); err != nil {
+		t.Fatal(err)
+	}
+	if err := journal.Refresh(); err == nil || !strings.Contains(err.Error(), "conflicting receipt journal id") {
+		t.Fatalf("conflicting refresh error=%v", err)
+	}
+}