feat(server): Phase 4 — gRPC proto extensions + soft-delete + reaper

[thebtf]

Summary

Phase 4 of the engram v4.4.0 loom integration train. Adds the server-side infrastructure that the daemon-side serverevents bridge (Phase 5) will consume: two new gRPC RPCs, an HTTP DELETE /api/projects/{id} endpoint, an in-process event bus, a DB migration, and an hourly soft-delete reaper job.

Implements spec FR-6 (SyncProjectState RPC), FR-7 (ProjectEvents streaming RPC), FR-8 (DELETE endpoint + soft-delete + lifecycle event source) + C3 clarification (soft-delete with 30-day retention window) + US2 (P1 project removal end-to-end).

This is the largest PR in the v4.4.0 train (~1,700 LOC production + tests).

What ships

Proto layer (T021 + T022)

• proto/engram/v1/engram.proto — appended SyncProjectState unary RPC + ProjectEvents server-streaming RPC + 4 new message types (SyncProjectStateRequest/Response, ProjectEventsRequest, ProjectEvent) + ProjectEventType enum. Existing 3 RPCs (CallTool, Initialize, Ping) byte-identical — engramcore contract preserved.
• proto/engram/v1/engram.pb.go + engram_grpc.pb.go — regenerated via protoc.

Database layer (T023)

• Migration 082_projects_lifecycle (NOT 025 as plan.md assumed — real highest engram migration is 081; plan.md spec was based on a stale semantic-search result). Adds removed_at TIMESTAMPTZ NULL + last_heartbeat TIMESTAMPTZ DEFAULT NOW() columns to the projects table. Idempotent IF NOT EXISTS guards. Rollback drops both columns.

In-process event bus (T024 + T025)

• internal/worker/projectevents/ — new package with Bus + Event types. Synchronous fan-out, panic-recovered per subscriber (matches loom.EventBus pattern). Subscribe(handler) func() returns unsubscribe closure. Tests cover concurrent emit, unsubscribe idempotency, panic recovery.

HTTP endpoint (T026 through T029)

• internal/worker/handlers_projects.go — handleDeleteProject method on Service. Extracts {id} from chi.URLParam, validates non-empty + hash shape, UPDATEs projects SET removed_at = NOW() WHERE id = $1 AND removed_at IS NULL inside a transaction, emits project_removed event on success, returns 200 {id, removed_at} on success / 400 malformed / 404 not found (or already deleted) / 500 DB error.
• Route r.Delete("/api/projects/{id}", s.handleDeleteProject) mounted in the require-ready chi.Group alongside existing admin routes — inherits ApiKeyAuth Bearer token middleware.
• Service struct gains eventBus *projectevents.Bus field, initialised in NewService, passed through to handlers via receiver.

gRPC handlers (T030 through T033)

• internal/grpcserver/sync_project_state.go — implements FR-6. Input validation: ≤10 000 IDs → INVALID_ARGUMENT, non-empty client_id → INVALID_ARGUMENT. Queries projects table for the union of input IDs. Returns removed[] for IDs with non-NULL removed_at OR not in table. Returns unknown[] for IDs with last_heartbeat > 24h AND removed_at IS NULL. UPDATEs last_heartbeat = NOW() for every reported ID as a side effect. Returns server_time_unix_ms for clock-skew detection.
• internal/grpcserver/project_events.go — implements FR-7. Server-streaming handler. Subscribes to projectevents.Bus on entry, forwards each Event to the gRPC stream. Returns OUT_OF_RANGE if since_event_id is non-empty (replay reserved for v0.2.0). Unsubscribes cleanly on ctx cancel or Send error. Terminates with OK on engram-server shutdown.
• Both handlers tested via bufconn fake server with 9 total test cases.

Soft-delete visibility audit (T034)

• Audited internal/search/, internal/worker/handlers_* for project-scoped queries via semantic codebase search. Two queries updated to filter removed_at IS NULL:
  • internal/worker/handlers_data.go:1179 — getProjectDisplayName() lookup
  • internal/db/gorm/project_store.go:90 — ResolveProjectID() legacy alias lookup
• Intentionally not filtered (correct behavior documented): handleDeleteProject UPDATE (needs to find the live row to mark deleted), SyncProjectState SELECT (needs to see soft-deleted rows to return them as removed[]), reaper DELETE (hard-deletes the soft-deleted rows), migration scripts.

Reaper job (T035 + T036 + T037)

• internal/worker/reaper/reaper.go — hourly soft-delete cleanup job. Reaper struct with Start(ctx) and Stop(). time.NewTicker(1 * time.Hour) loop. Each tick executes DELETE FROM projects WHERE removed_at < NOW() - retention_interval, where retention is driven by ENGRAM_PROJECT_RETENTION_DAYS env var (default 30). Respects ctx.Done for graceful shutdown. Reaper started in Service.Start after DB is ready, stopped in shutdown path.

CASCADE-DELETE FK AUDIT (key finding): engram has no ON DELETE CASCADE foreign keys referencing projects.id. All project-scoped tables (observations, sdk_sessions, injection_log, issues, project_settings, patterns, embeddings, collections, memory_blocks) carry denormalized project TEXT columns managed by their own maintenance workflows. The reaper therefore deletes only from the projects table — child data is NOT cascade-deleted and remains visible until the respective tables' own cleanup runs. This is a pre-existing architectural choice in engram; it's documented here so future maintainers don't assume reaper does a deep cascade.

10 atomic commits

Commit	Task	Description
aac4a47	T021	proto RPCs + messages added
6a80085	T022	regenerated pb.go + grpc.pb.go
1583734	T023	migration 082_projects_lifecycle
12cb0db	T024 + T025	projectevents.Bus + tests
7f50812	T026	handlers_projects.go handler (75 LOC)
45bc756	T027	handlers_projects_test.go (302 LOC, 7 tests)
4c911fa	T028 + T029 + T037	Service.eventBus field + DELETE route + reaper start
416baaa	T030	grpcserver/sync_project_state.go (144 LOC)
407c057	T031	sync_project_state_test.go (202 LOC, 5 tests)
14c9370	T032	grpcserver/project_events.go (102 LOC)
64ea59b	T033	project_events_test.go (202 LOC, 4 bufconn tests)
a70db66	T034	soft-delete visibility audit (2 files)
c4160b4	T035	reaper/reaper.go (148 LOC)
8f4ab4b	T036	reaper/reaper_test.go (173 LOC, 4 tests)

Verification

• go build ./... — clean
• go vet ./... — clean
• go test ./... -count=1 — all 48 packages PASS, zero regressions
• Migration 082 idempotent + rollback verified locally
• DB-requiring tests (T027, T031, T036) skip cleanly when DATABASE_DSN is unset; validation-only tests always run

Compatibility

• engramcore byte-identity preserved — NFR-5 contract test still passes. Existing 3 gRPC RPCs unchanged.
• Plugin wire format unchanged — daemon-server protocol additions do not affect the client plugin.
• Zero new CGO dependencies — NFR-6 holds. Only stdlib + existing engram deps.
• Constitution §15 — server + plugin will unify to v4.4.0 in Phase 6 ceremony after all PRs merge.

What does NOT change

• engramcore tenant (internal/handlers/engramcore/*) — untouched
• loom tenant (internal/handlers/loom/*) — untouched (owned by PR feat(loom): Phase B-1 plumbing tenant — loom v0.1.0 integration #167/feat(obs,loom): Phase 2 — obs.MeterFor() helper + loom OTel wiring #168/feat(loom): Phase 3 — 4 MCP tools + CLI worker with allowlist #169 which already merged)
• daemon side — this PR is server-side only (cmd/engram-server/, internal/worker/, internal/grpcserver/, internal/db/gorm/, proto/)
• Dashboard UI — out of scope; operators use curl/httpie until v4.5+

Related

• Spec: .agent/specs/loom-integration/spec.md FR-6/7/8 + C3 + US2 (gitignored)
• Plan: .agent/specs/loom-integration/plan.md §Phase 4 (gitignored)
• Tasks: .agent/specs/loom-integration/tasks.md T021-T037 + G004 (gitignored)
• Proto schema source: .agent/specs/loom-integration/proto-extensions.md (gitignored)
• Previous PRs in v4.4.0 train: feat(loom): Phase B-1 plumbing tenant — loom v0.1.0 integration #167 (plumbing) merged, feat(obs,loom): Phase 2 — obs.MeterFor() helper + loom OTel wiring #168 (OTel meter) merged, feat(loom): Phase 3 — 4 MCP tools + CLI worker with allowlist #169 (tools) merging now
• Next in train: PR loom-serverevents (daemon bridge consuming this PR's new RPCs) — depends on this PR merging for regenerated proto stubs
• Final: v4.4.0 tag ceremony after all 5 PRs merge

[coderabbitai]

Warning

Rate limit exceeded

@thebtf has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 7 minutes and 6 seconds before requesting another review.

Your organization is not enrolled in usage-based pricing. Contact your admin to enable usage-based pricing to continue reviews beyond the rate limit, or try again in 7 minutes and 6 seconds.

⌛ How to resolve this issue?

After the wait time has elapsed, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

We recommend that you space out your commits to avoid hitting the rate limit.

🚦 How do rate limits work?

CodeRabbit enforces hourly rate limits for each developer per organization.

Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout.

Please see our FAQ for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 123328ec-d2c3-463b-a875-d24488e0897e

📥 Commits

Reviewing files that changed from the base of the PR and between ff57fec and 8f4ab4b.

⛔ Files ignored due to path filters (2)

• proto/engram/v1/engram.pb.go is excluded by !**/*.pb.go
• proto/engram/v1/engram_grpc.pb.go is excluded by !**/*.pb.go

📒 Files selected for processing (17)

• internal/db/gorm/migrations.go
• internal/db/gorm/models.go
• internal/db/gorm/project_store.go
• internal/grpcserver/project_events.go
• internal/grpcserver/project_events_test.go
• internal/grpcserver/server.go
• internal/grpcserver/sync_project_state.go
• internal/grpcserver/sync_project_state_test.go
• internal/worker/handlers_data.go
• internal/worker/handlers_projects.go
• internal/worker/handlers_projects_test.go
• internal/worker/projectevents/bus.go
• internal/worker/projectevents/bus_test.go
• internal/worker/reaper/reaper.go
• internal/worker/reaper/reaper_test.go
• internal/worker/service.go
• proto/engram/v1/engram.proto

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/engram-server-proto-b0

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[thebtf]

@coderabbitai review focus:security, performance, best-practices

[gemini-code-assist]

Code Review

This pull request implements project lifecycle management, including soft-deletion, heartbeat tracking, and real-time event notifications. Key changes include database migrations for removed_at and last_heartbeat columns, the addition of gRPC endpoints for project state synchronization and event streaming, and an HTTP handler for soft-deleting projects. It also introduces an in-process event bus and a periodic reaper job to hard-delete projects after a configurable retention period. Feedback was provided regarding a date typo in the reaper's audit documentation.

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
echo "=== handlers_projects.go ==="
fd -e go -g 'handlers_projects.go' --exec cat {}

echo "=== sync_project_state.go ==="
fd -e go -g 'sync_project_state.go' --exec cat {}

echo "=== project_events.go (grpcserver) ==="
fd -e go -g 'project_events.go' --exec cat {}

echo "=== reaper.go ==="
fd -e go -g 'reaper.go' --exec cat {}

echo "=== projectevents bus ==="
fd -e go -p 'projectevents' --exec cat {}

echo "=== project_store.go ==="
fd -e go -g 'project_store.go' --exec cat {}

echo "=== handlers_data.go lines around 1179 ==="
fd -e go -g 'handlers_data.go' --exec sed -n '1160,1200p' {}

echo "=== Service struct + NewService ==="
fd -e go -g 'service.go' -p 'worker' --exec cat {}

echo "=== migration 082 ==="
fd -e sql -g '*082*' --exec cat {}
fd -e go -g '*082*' --exec cat {}

Length of output: 25068

---

🏁 Script executed:

#!/bin/bash
echo "=== Service struct (find service.go in worker) ==="
find . -type f -name 'service.go' | xargs grep -l 'eventBus\|Service struct' 2>/dev/null | head -5 | xargs cat

echo "=== Migration 082 ==="
find . -type f | grep -i '082' | head -10 | xargs cat 2>/dev/null

echo "=== ValidateProjectName ==="
rg -n 'func ValidateProjectName' -A 20

echo "=== Route registration (chi Delete route) ==="
rg -n 'handleDeleteProject\|r\.Delete' -C 3

echo "=== grpcserver Server struct ==="
rg -n 'type Server struct' -A 10 --type go

echo "=== grpcserver token auth middleware ==="
rg -n 'token\|ApiKeyAuth\|Bearer\|Authorization' --type go -l

echo "=== ProjectEvents auth check ==="
rg -n 'token\|auth\|intercept' internal/grpcserver/ -A 3

echo "=== nowUnixMs unused? ==="
rg -n 'nowUnixMs' --type go

echo "=== retentionDays called per tick ==="
rg -n 'retentionDays' --type go -A 3

echo "=== ANY(?) postgres driver compatibility ==="
rg -n 'ANY\(\?\)' --type go -B 2 -A 2

echo "=== Stop called without Start? done channel block ==="
rg -n 'func.*Stop\(\)' internal/worker/reaper/ -A 15

echo "=== sync_project_state unknown includes caller's own IDs? ==="
rg -n 'unknown\|staleIDs' internal/grpcserver/sync_project_state.go

echo "=== grpc server unary interceptors ==="
rg -n 'UnaryInterceptor\|StreamInterceptor\|grpc.NewServer' --type go -A 5

Length of output: 157253

---

@thebtf Here is the focused security, performance, and best-practices review of this PR.

---

🔴 Security

1. ANY(?) without pq.Array() — runtime query failure

internal/grpcserver/sync_project_state.go, Lines 64 & 90

The two new raw queries pass a bare []string to PostgreSQL's ANY(?) operator:

// Line 64
Raw("SELECT id, removed_at FROM projects WHERE id = ANY(?)", localIDs).

// Line 90
Exec("UPDATE projects SET last_heartbeat = ? WHERE id = ANY(?) AND removed_at IS NULL", now, localIDs)

The existing codebase wraps Go slices in pq.Array() for this exact pattern (see internal/db/gorm/injection_log_store.go line 52: pq.Array(observationIDs)). Without the wrapper, lib/pq cannot encode a []string as a PostgreSQL array literal, causing a runtime driver error. This will silently succeed in unit tests (which skip when DATABASE_DSN is unset) but fail against a real database.

Fix:

import "github.com/lib/pq"

// Line 64
Raw("SELECT id, removed_at FROM projects WHERE id = ANY(?)", pq.Array(localIDs)).

// Line 90
Exec("UPDATE projects SET last_heartbeat = ? WHERE id = ANY(?) AND removed_at IS NULL",
    now, pq.Array(localIDs))

---

2. SyncProjectState — global information disclosure in unknown[]

internal/grpcserver/sync_project_state.go, Lines 96–105

The stale/orphan query is unbounded and not scoped to the caller's reported IDs or client_id:

s.db.WithContext(ctx).
    Raw(
        "SELECT id FROM projects WHERE removed_at IS NULL AND (last_heartbeat IS NULL OR last_heartbeat < ?)",
        staleThreshold,
    ).
    Pluck("id", &staleIDs)

Any authenticated daemon receives the full list of all stale projects across the entire server, including those belonging to unrelated daemons. This is an information-disclosure issue: daemon A can enumerate project IDs it has no relationship to.

Recommended fix: scope the query to IDs the caller has previously reported, or introduce a client_id-to-project ownership model. At minimum, add a LIMIT to cap the response size:

"SELECT id FROM projects WHERE removed_at IS NULL AND (last_heartbeat IS NULL OR last_heartbeat < ?) LIMIT 1000"

---

3. Reaper.Stop() deadlocks if called before Start()

internal/worker/reaper/reaper.go, Line 91–99

func (r *Reaper) Stop() {
    select {
    case <-r.stop:   // already closed
    default:
        close(r.stop)
    }
    <-r.done  // ← blocks forever if Start() was never called
}

r.done is only closed by the goroutine launched in Start(). If Stop() is ever called on an unstarted Reaper (e.g., in test teardown or after a startup error), it deadlocks permanently. The service Shutdown() path relies on context cancellation rather than Stop(), which avoids the issue in production — but the Stop() API is publicly exposed and unsafe.

Fix:

func (r *Reaper) Stop() {
    select {
    case <-r.stop:
    default:
        close(r.stop)
    }
    select {
    case <-r.done:
    default:
        // Start() was never called; nothing to wait for.
    }
}

---

🟡 Performance

4. retentionDays() performs an os.Getenv syscall on every hourly purge

internal/worker/reaper/reaper.go, Line 108

func (r *Reaper) purge(ctx context.Context) {
    retention := time.Duration(retentionDays()) * 24 * time.Hour  // env read every tick

retentionDays() re-reads ENGRAM_PROJECT_RETENTION_DAYS from the environment on every invocation. For an hourly job this is negligible in absolute terms, but it's inconsistent with the expectation that Start() logs the retention once. More importantly, it means the retention window can silently change between ticks without any log, making it harder to audit.

Fix: read and store the value once in New() or Start():

type Reaper struct {
    db            *gorm.DB
    stop          chan struct{}
    done          chan struct{}
    retentionDays int  // resolved once at construction
}

func New(db *gorm.DB) *Reaper {
    return &Reaper{
        db:            db,
        stop:          make(chan struct{}),
        done:          make(chan struct{}),
        retentionDays: retentionDays(),
    }
}

5. Unbounded unknown[] stale query — no LIMIT

Already noted in finding #2. For a table with thousands of projects, the SELECT id FROM projects WHERE removed_at IS NULL AND (last_heartbeat IS NULL OR last_heartbeat < ?) scan returns an unbounded result set included in every SyncProjectState response. Add a LIMIT aligned with maxSyncProjectIDs:

SELECT id FROM projects
WHERE removed_at IS NULL
  AND (last_heartbeat IS NULL OR last_heartbeat < ?)
LIMIT 10000

---

🔵 Best Practices

6. nowUnixMs() is dead code

internal/grpcserver/project_events.go, Lines 99–101

// nowUnixMs returns the current time as Unix milliseconds.
func nowUnixMs() int64 {
    return time.Now().UnixMilli()
}

This function is defined but never called anywhere in the codebase. It will produce a go vet / linter warning and adds noise. Remove it or replace the inline time.Now().UnixMilli() calls in busEventToProto with it for consistency — but don't leave it unused.

---

7. Heartbeat update errors silently discarded with no telemetry

internal/grpcserver/sync_project_state.go, Lines 87–93

// Ignore errors — heartbeat is best-effort.
if len(localIDs) > 0 {
    s.db.WithContext(ctx).
        Exec("UPDATE projects SET last_heartbeat = ? WHERE id = ANY(?) AND removed_at IS NULL", now, localIDs)
}

While best-effort is a legitimate design choice, completely discarding the error—even at debug level—makes it impossible to distinguish "DB is down" from "IDs don't exist." A single log.Debug().Err(result.Error) preserves the intent while improving diagnosability.

---

8. PurgeOnce() always returns nil — swallows purge errors from callers

internal/worker/reaper/reaper.go, Lines 143–149

func (r *Reaper) PurgeOnce(ctx context.Context) error {
    if r.db == nil {
        return fmt.Errorf("reaper: db is nil")
    }
    r.purge(ctx)  // ← errors logged internally but not returned
    return nil
}

purge() logs errors via log.Error() but PurgeOnce always returns nil. Integration tests using PurgeOnce cannot distinguish a successful purge from a failed one, defeating the purpose of the error return type. Fix:

func (r *Reaper) PurgeOnce(ctx context.Context) error {
    if r.db == nil {
        return fmt.Errorf("reaper: db is nil")
    }
    return r.purge(ctx)  // propagate
}

// Change purge() signature to return error
func (r *Reaper) purge(ctx context.Context) error { ... }

---

9. Redundant inner len(localIDs) > 0 guard

internal/grpcserver/sync_project_state.go, Lines 80–93

The heartbeat UPDATE block at line 80 is already inside the outer if len(localIDs) > 0 block (line 47). The inner guard is dead:

// Outer guard at line 47 already ensures len(localIDs) > 0
if len(localIDs) > 0 {   // ← redundant inner check
    s.db.WithContext(ctx).Exec(...)
}

Remove the inner check to reduce cognitive noise.

---

Summary Table

#	Area	Severity	File	Finding
1	Security	🔴 Critical	sync_project_state.go	ANY(?) without pq.Array() — runtime DB failure
2	Security	🔴 High	sync_project_state.go	unknown[] is a global scan — information disclosure + unbounded response
3	Security	🟠 Medium	reaper/reaper.go	Stop() deadlocks if called before Start()
4	Performance	🟡 Low	reaper/reaper.go	os.Getenv on every hourly tick
5	Performance	🟡 Low	sync_project_state.go	No LIMIT on stale project scan
6	Best Practice	🔵 Low	project_events.go	nowUnixMs() dead code — remove
7	Best Practice	🔵 Low	sync_project_state.go	Heartbeat errors silently discarded
8	Best Practice	🔵 Low	reaper/reaper.go	PurgeOnce() always returns nil
9	Best Practice	🔵 Info	sync_project_state.go	Redundant inner len(localIDs) > 0 guard

Finding #1 (pq.Array()) is the most urgent — it will cause SyncProjectState to fail against any real PostgreSQL deployment.