fix(sync): bound OpenCode background work to the changed session batch

Makefile

@@ -14,7 +14,11 @@ GOLANGCI_LINT_VERSION ?= v2.11.4
 # Isolate each checkout from stale sibling-worktree fixes and issue positions.
 GOLANGCI_LINT_CACHE ?= $(CURDIR)/.golangci-cache
 export GOLANGCI_LINT_CACHE
+ifeq ($(findstring NT-,$(shell uname -s 2>/dev/null)),NT-)
+CUSTOM_GCL := ./custom-gcl.exe
+else
 CUSTOM_GCL := ./custom-gcl
+endif
 PRICING_SNAPSHOT_FILE := internal/pricing/snapshot/litellm_snapshot.json.gz
 
 # sqlite-vec's cgo bindings #include "sqlite3.h". Without an override the
@@ -424,7 +428,7 @@ nilaway-golangci-build:
 		exit 1; \
 	fi
 	@unset_args=$$(git rev-parse --local-env-vars 2>/dev/null | sed 's/^/-u /' | tr '\n' ' '); \
-	env $$unset_args GOFLAGS=-buildvcs=false \
+	env $$unset_args GOFLAGS=-buildvcs=false GOTOOLCHAIN=$(shell go env GOVERSION) \
 		golangci-lint custom --version "$(GOLANGCI_LINT_VERSION)" --name custom-gcl
 
 # Run NilAway through the custom golangci-lint module plugin.
@@ -461,13 +465,17 @@ nilaway: pricing-snapshot ensure-embed-dir nilaway-golangci-build
 	root=$$(pwd); \
 	dirs=$$(go list -f '{{.Dir}}' ./...); \
 	pkgs=$$(for dir in $$dirs; do \
+		case "$$(uname -s 2>/dev/null)" in \
+			*_NT-*) dir=$$(cygpath -u "$$dir");; \
+		esac; \
 		if [ "$$dir" = "$$root" ]; then \
 			printf '%s\n' "."; \
 		else \
 			printf '%s\n' "./$${dir#$$root/}"; \
 		fi; \
 	done); \
 	if [ -z "$$pkgs" ]; then echo "nilaway: no packages to lint" >&2; exit 1; fi; \
+	count=$$(printf '%s\n' "$$pkgs" | awk 'END { print NR }'); \
 	NILAWAY_OUT_DIR=$$(mktemp -d); \
 	export NILAWAY_OUT_DIR; \
 	trap 'rm -f "$$NILAWAY_OUT_DIR"/*; rmdir "$$NILAWAY_OUT_DIR"' EXIT HUP INT TERM; \
@@ -487,8 +495,10 @@ nilaway: pricing-snapshot ensure-embed-dir nilaway-golangci-build
 		exit $$status' sh; \
 	rc=$$?; \
 	set -e; \
-	ls "$$NILAWAY_OUT_DIR" | sort -n | while IFS= read -r n; do \
-		cat "$$NILAWAY_OUT_DIR/$$n"; \
+	n=1; \
+	while [ "$$n" -le "$$count" ]; do \
+		if [ -f "$$NILAWAY_OUT_DIR/$$n" ]; then cat "$$NILAWAY_OUT_DIR/$$n"; fi; \
+		n=$$((n + 1)); \
 	done; \
 	exit $$rc
 

cmd/agentsview/archive_write_backend.go

@@ -106,10 +106,12 @@ func newArchivePushUnwatchedPoller(
 		return hooks.newUnwatchedPoller(ctx, engine)
 	}
 	ticker := time.NewTicker(unwatchedPollInterval)
-	return newUnwatchedPollCoordinatorWithTicks(
+	poller := newUnwatchedPollCoordinatorWithTicks(
 		ctx, engine, ticker.C, ticker.Stop, func(work func()) { work() }, nil,
 		time.Now, time.After,
 	)
+	poller.DisableBoundedCoverage()
+	return poller
 }
 
 func startArchivePushWatcher(

cmd/agentsview/bounded_coverage_integration_test.go

@@ -0,0 +1,212 @@
+package main
+
+import (
+	"database/sql"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.kenn.io/agentsview/internal/dbtest"
+	"go.kenn.io/agentsview/internal/parser"
+	agentsync "go.kenn.io/agentsview/internal/sync"
+)
+
+// TestBoundedCoverageCoordinatorCardinality uses the production coordinator,
+// Engine resolver, journal drain, and source-application seam. Only the
+// journal's changed row is observed for both archive cardinalities.
+func TestBoundedCoverageCoordinatorCardinality(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping bounded coverage cardinality integration")
+	}
+	for _, mode := range []struct {
+		name string
+	}{
+		{name: "native"},
+		{name: "degraded"},
+	} {
+		for _, sessions := range []int{10, 5000} {
+			t.Run(fmt.Sprintf("%s_sessions_%d", mode.name, sessions), func(t *testing.T) {
+				root := t.TempDir()
+				dbPath := filepath.Join(root, "opencode.db")
+				journal, err := sql.Open("sqlite3", dbPath)
+				require.NoError(t, err)
+				t.Cleanup(func() { _ = journal.Close() })
+				_, err = journal.Exec("PRAGMA wal_autocheckpoint=0")
+				require.NoError(t, err)
+				_, err = journal.Exec(boundedCoverageFixtureSchema)
+				require.NoError(t, err)
+				var journalMode string
+				err = journal.QueryRow("PRAGMA journal_mode=WAL").Scan(&journalMode)
+				require.NoError(t, err)
+				require.Equal(t, "wal", journalMode)
+				_, err = journal.Exec("INSERT INTO project (id, worktree, time_updated) VALUES ('proj', ?, 1)", root)
+				require.NoError(t, err)
+				for i := range sessions {
+					id := fmt.Sprintf("ses%05d", i)
+					_, err = journal.Exec(
+						"INSERT INTO session (id, project_id, time_created, time_updated) VALUES (?, 'proj', 1, 1)", id,
+					)
+					require.NoError(t, err)
+				}
+				_, err = journal.Exec(`INSERT INTO message
+			(id, session_id, data, time_created, time_updated)
+			VALUES ('msg-0', 'ses00000', '{"role":"assistant"}', 1, 1)`)
+				require.NoError(t, err)
+				_, err = journal.Exec(`INSERT INTO part
+			(id, session_id, message_id, data, time_created, time_updated)
+			VALUES ('part-0', 'ses00000', 'msg-0',
+			'{"type":"text","content":"changed"}', 1, 1)`)
+				require.NoError(t, err)
+				archive := dbtest.OpenTestDB(t)
+				engine := agentsync.NewEngine(archive, agentsync.EngineConfig{
+					AgentDirs: map[parser.AgentType][]string{parser.AgentOpenCode: {root}},
+					Machine:   "local",
+				})
+				t.Cleanup(engine.Close)
+				ticks := make(chan time.Time)
+				coordinator := newUnwatchedPollCoordinatorWithTicks(
+					t.Context(), engine, ticks, func() {}, func(func()) {}, nil,
+					time.Now, time.After,
+				)
+				t.Cleanup(coordinator.Stop)
+				var rows, applied int
+				coordinator.onBoundedCoveragePage = func(result parser.OpenCodeFeedResult) {
+					rows += result.RowsRead
+				}
+				coordinator.onBoundedCoverageApply = func(stats agentsync.SyncStats) {
+					applied += stats.Synced
+				}
+				roots := []agentsync.BoundedCoverageRoot{{Agent: parser.AgentOpenCode, Root: root}}
+				bindings, err := engine.BoundedCoverageBindings(t.Context(), roots)
+				require.NoError(t, err)
+				eventTypes := []string{"message.updated", "message.part.updated", "session.updated"}
+				for i, eventType := range eventTypes {
+					payload := "{}"
+					if eventType == "message.updated" {
+						payload = `{"sessionID":"ses00000","info":{"id":"msg-0","sessionID":"ses00000","role":"user"}}`
+					}
+					_, err = journal.Exec(`INSERT INTO event
+					(id, aggregate_id, seq, type, data)
+					VALUES (?, 'ses00000', ?, ?, ?)`, fmt.Sprintf("event-before-%d", i), i+1, eventType, payload)
+					require.NoError(t, err)
+				}
+				if mode.name == "native" {
+					_, err = coordinator.AdmitBoundedCoverage(t.Context(), bindings, true)
+					require.NoError(t, err)
+				} else {
+					require.NoError(t, coordinator.AddObligation(pollingObligation{
+						Key: "degraded", Scopes: []pollingScope{{Agent: parser.AgentOpenCode, Root: root}},
+					}))
+				}
+				for i, eventType := range eventTypes {
+					payload := "{}"
+					if eventType == "message.updated" {
+						payload = `{"sessionID":"ses00000","info":{"id":"msg-0","sessionID":"ses00000","role":"user"}}`
+					}
+					_, err = journal.Exec(`INSERT INTO event
+					(id, aggregate_id, seq, type, data)
+					VALUES (?, 'ses00000', ?, ?, ?)`, fmt.Sprintf("event-after-%d", i), i+4, eventType, payload)
+					require.NoError(t, err)
+				}
+				walInfo, err := os.Stat(dbPath + "-wal")
+				require.NoError(t, err)
+				require.Greater(t, walInfo.Size(), int64(32),
+					"the measured mutation must retain WAL frames beyond its header")
+				require.NoError(t, coordinator.pollBoundedCoverageOnce(t.Context()))
+				_, err = archive.GetSession(t.Context(), "ses00000")
+				require.NoError(t, err)
+				t.Logf("bounded_admission mode=%s sessions=%d event_types=%s observed_journal_rows=%d applied_sources=%d source=%s wal_bytes=%d", mode.name, sessions, strings.Join(eventTypes, ","), rows, applied, bindings[0].PhysicalDBPath, walInfo.Size())
+				require.Equal(t, len(eventTypes)*2, rows,
+					"native and degraded production admission must retain every producer event")
+				require.Equal(t, 1, applied)
+				require.LessOrEqual(t, rows, parser.OpenCodeCoverageMaxRows)
+			})
+		}
+	}
+}
+
+func TestBoundedCoverageLeaseRejectsReplacedDatabase(t *testing.T) {
+	root := t.TempDir()
+	dbPath := filepath.Join(root, "opencode.db")
+	journal, err := sql.Open("sqlite3", dbPath)
+	require.NoError(t, err)
+	_, err = journal.Exec(boundedCoverageFixtureSchema)
+	require.NoError(t, err)
+	require.NoError(t, journal.Close())
+
+	archive := dbtest.OpenTestDB(t)
+	engine := agentsync.NewEngine(archive, agentsync.EngineConfig{
+		AgentDirs: map[parser.AgentType][]string{parser.AgentOpenCode: {root}},
+		Machine:   "local",
+	})
+	t.Cleanup(engine.Close)
+	bindings, err := engine.BoundedCoverageBindings(t.Context(), []agentsync.BoundedCoverageRoot{{Agent: parser.AgentOpenCode, Root: root}})
+	require.NoError(t, err)
+	require.Len(t, bindings, 1)
+	bindings[0].Generation = 1
+	lease, err := engine.AdmitBoundedCoverageLease(t.Context(), bindings[0])
+	require.NoError(t, err)
+	_, err = engine.TransitionBoundedCoverageRequest(t.Context(), lease, nil, lease.AdmissionCheckpoint, true)
+	require.NoError(t, err)
+
+	backup := filepath.Join(root, "opencode.old.db")
+	require.NoError(t, os.Rename(dbPath, backup))
+	replacement, err := sql.Open("sqlite3", dbPath)
+	require.NoError(t, err)
+	_, err = replacement.Exec(boundedCoverageFixtureSchema)
+	require.NoError(t, err)
+	require.NoError(t, replacement.Close())
+
+	_, err = engine.TransitionBoundedCoverageRequest(t.Context(), lease, nil, lease.AdmissionCheckpoint, false)
+	require.Error(t, err, "replacement must invalidate the old physical lease before commit")
+}
+
+func TestBoundedCoverageBindingsDeduplicateSymlinkedRoots(t *testing.T) {
+	root := t.TempDir()
+	dbPath := filepath.Join(root, "opencode.db")
+	db, err := sql.Open("sqlite3", dbPath)
+	require.NoError(t, err)
+	_, err = db.Exec(boundedCoverageFixtureSchema)
+	require.NoError(t, err)
+	require.NoError(t, db.Close())
+
+	aliasParent := t.TempDir()
+	alias := filepath.Join(aliasParent, "alias")
+	if err := os.Symlink(root, alias); err != nil {
+		t.Skipf("directory symlinks unavailable: %v", err)
+	}
+
+	archive := dbtest.OpenTestDB(t)
+	engine := agentsync.NewEngine(archive, agentsync.EngineConfig{
+		AgentDirs: map[parser.AgentType][]string{parser.AgentOpenCode: {root, alias}},
+		Machine:   "local",
+	})
+	defer engine.Close()
+	bindings, err := engine.BoundedCoverageBindings(t.Context(), []agentsync.BoundedCoverageRoot{
+		{Agent: parser.AgentOpenCode, Root: root},
+		{Agent: parser.AgentOpenCode, Root: alias},
+	})
+	require.NoError(t, err)
+	require.Len(t, bindings, 1,
+		"lexical aliases of one physical database must share one coverage binding")
+	assert.Equal(t, filepath.Clean(dbPath), bindings[0].DBPath)
+	assert.Equal(t, filepath.Clean(root), bindings[0].Scope)
+}
+
+const boundedCoverageFixtureSchema = `
+CREATE TABLE project (id TEXT PRIMARY KEY, worktree TEXT NOT NULL, time_updated INTEGER NOT NULL);
+CREATE TABLE session (id TEXT PRIMARY KEY, project_id TEXT NOT NULL, parent_id TEXT, title TEXT, time_created INTEGER NOT NULL, time_updated INTEGER NOT NULL);
+CREATE TABLE message (id TEXT PRIMARY KEY, session_id TEXT NOT NULL, data TEXT NOT NULL, time_created INTEGER NOT NULL, time_updated INTEGER NOT NULL);
+CREATE TABLE part (id TEXT PRIMARY KEY, session_id TEXT NOT NULL, message_id TEXT NOT NULL, data TEXT NOT NULL, time_created INTEGER NOT NULL, time_updated INTEGER NOT NULL);
+CREATE INDEX message_session_time_created_id_idx ON message (session_id, time_created, id);
+CREATE INDEX part_session_idx ON part (session_id);
+CREATE INDEX part_message_id_id_idx ON part (message_id, id);
+CREATE TABLE event (id TEXT NOT NULL PRIMARY KEY, aggregate_id TEXT NOT NULL, seq INTEGER NOT NULL, type TEXT NOT NULL, data BLOB NOT NULL);
+CREATE TABLE event_sequence (id TEXT NOT NULL PRIMARY KEY, owner_id TEXT);
+`