fix(control_plane/memory): address P1 review issues — async interface, privacy, authority

Fix four P1 issues from code review:

1. Async/sync interface mismatch: MemoryStore interface now declares
   async methods (Promise return types). FileMemoryStore and
   MemoryStoreInMemory both implement the async contract. All consumers
   (discovery, resume, handoff, projection) updated to await store calls.

2. adoptMemory self-link bug: adoptMemory now requires a distinct
   successor_memory_id and validates that it exists, belongs to the same
   goal, and its supersedes field points back to the source. Self-links
   (source == successor) are rejected with EffectRuntimeRequestError.

3. No privacy enforcement: discoverMemories now accepts an optional
   MemoryViewerContext. Session-scoped records from other sessions are
   filtered out when a viewer context is present. createMemoryRecord
   defaults public_safe/redaction_verified to false (caller must opt in).
   Projection functions only include public_safe records in summaries
   and handoff rows.

4. Handoff authority is just text: composeHandoffWithMode now validates
   that soft_claim/hard_lease modes require handoff-availability records
   that are certified public_safe. Legacy mode remains permissive.
   Authority labels are validated against record metadata, not just
   returned as opaque strings.

Test count: 64 → 73 (added tests for new validation paths).

Signed-off-by: xiaodeshi <xiaodeshi@users.noreply.github.com>
Signed-off-by: Xiao Deshi <xiaods@gmail.com>

Author: xiaods

loopx/control_plane/memory/index.ts

@@ -42,6 +42,7 @@ export {
   discoverMemories,
   getSupersedeChain,
   type MemoryDiscoveryFilter,
+  type MemoryViewerContext,
 } from "./memory_discovery.ts";
 
 export {

loopx/control_plane/memory/memory_discovery.ts

@@ -11,6 +11,11 @@ import type { MemoryStore } from "./memory_store.ts";
 // Discovery filters
 // ---------------------------------------------------------------------------
 
+export interface MemoryViewerContext {
+  readonly session_ref: string;
+  readonly agent_id: string;
+}
+
 export interface MemoryDiscoveryFilter {
   readonly goal_id?: string;
   readonly status?: MemoryStatus;
@@ -20,17 +25,18 @@ export interface MemoryDiscoveryFilter {
   readonly session_ref?: string;
   readonly semantic_query?: string;
   readonly limit?: number;
+  readonly viewer?: MemoryViewerContext;
 }
 
 // ---------------------------------------------------------------------------
 // Discovery API
 // ---------------------------------------------------------------------------
 
-export function discoverMemories(
+export async function discoverMemories(
   store: MemoryStore,
   filter: MemoryDiscoveryFilter = {},
-): MemoryRecord[] {
-  const all = store.readAll();
+): Promise<MemoryRecord[]> {
+  const all = await store.readAll();
   let results = all;
 
   if (filter.goal_id !== undefined) {
@@ -61,6 +67,18 @@ export function discoverMemories(
     results = results.filter((r) => r.session_ref === filter.session_ref);
   }
 
+  // Privacy enforcement: session-scoped memories are only visible to
+  // the session that created them. When a viewer context is provided,
+  // session-private records from other sessions are excluded.
+  if (filter.viewer !== undefined) {
+    results = results.filter((r) => {
+      if (r.availability === "session") {
+        return r.session_ref === filter.viewer!.session_ref;
+      }
+      return true;
+    });
+  }
+
   if (filter.semantic_query !== undefined) {
     const query = filter.semantic_query.toLowerCase();
     results = results.filter((r) =>
@@ -83,34 +101,34 @@ export function discoverMemories(
   return results;
 }
 
-export function discoverLatestMemory(
+export async function discoverLatestMemory(
   store: MemoryStore,
   todo_id: string,
-): MemoryRecord | null {
-  const matches = discoverMemories(store, { todo_id, status: "active" });
+): Promise<MemoryRecord | null> {
+  const matches = await discoverMemories(store, { todo_id, status: "active" });
   return matches.length > 0 ? matches[0] : null;
 }
 
-export function discoverHandoffMemories(
+export async function discoverHandoffMemories(
   store: MemoryStore,
   goal_id: string,
-): MemoryRecord[] {
-  return discoverMemories(store, {
+): Promise<MemoryRecord[]> {
+  return (await discoverMemories(store, {
     goal_id,
     status: "active",
     availability: "handoff",
-  }).filter((r) => !r.superseded_by);
+  })).filter((r) => !r.superseded_by);
 }
 
 // ---------------------------------------------------------------------------
 // Supersede chain traversal
 // ---------------------------------------------------------------------------
 
-export function getSupersedeChain(
+export async function getSupersedeChain(
   store: MemoryStore,
   memory_id: string,
-): MemoryRecord[] {
-  const all = store.readAll();
+): Promise<MemoryRecord[]> {
+  const all = await store.readAll();
   const byId = new Map(all.map((r) => [r.memory_id, r]));
 
   // Walk backward to find the root

loopx/control_plane/memory/memory_handoff.ts

@@ -44,16 +44,18 @@ export interface HandoffMemoryWriteInput {
   readonly evidence_refs?: readonly string[];
   readonly supersedes?: string | null;
   readonly source_run_id?: string | null;
+  readonly public_safe?: boolean;
+  readonly redaction_verified?: boolean;
 }
 
 /**
  * Write a memory record during handoff preparation.
  * This is the "source session" side of the handoff flow.
  */
-export function writeHandoffMemory(
+export async function writeHandoffMemory(
   store: MemoryStore,
   input: HandoffMemoryWriteInput,
-): MemoryRecord {
+): Promise<MemoryRecord> {
   return store.append({
     goal_id: input.goal_id,
     session_ref: input.session_ref,
@@ -67,6 +69,8 @@ export function writeHandoffMemory(
     evidence_refs: input.evidence_refs ?? [],
     supersedes: input.supersedes ?? null,
     source_run_id: input.source_run_id ?? null,
+    public_safe: input.public_safe ?? false,
+    redaction_verified: input.redaction_verified ?? false,
   });
 }
 
@@ -91,23 +95,23 @@ export function buildMemoryHandoffNoteRef(
  * Discover handoff-available memories for a goal.
  * This is the "target session" side of the handoff flow.
  */
-export function discoverHandoffMemoriesForGoal(
+export async function discoverHandoffMemoriesForGoal(
   store: MemoryStore,
   goal_id: string,
-): MemoryRecord[] {
+): Promise<MemoryRecord[]> {
   return discoverHandoffMemories(store, goal_id);
 }
 
 /**
  * Resume from a handoff memory, producing a resume packet.
  * The target session decides whether to adopt.
  */
-export function resumeFromHandoffMemory(
+export async function resumeFromHandoffMemory(
   store: MemoryStore,
   memory_id: string,
   target_agent_id: string,
-): MemoryResumePacket {
-  const memory = store.readById(memory_id);
+): Promise<MemoryResumePacket> {
+  const memory = await store.readById(memory_id);
   if (!memory) {
     throw new EffectRuntimeRequestError(`handoff memory not found: ${memory_id}`);
   }
@@ -148,12 +152,32 @@ export interface HandoffCompositionResult {
  *   claim = adopt.
  * - hard_lease: memory transfer requires the target to hold the task lease;
  *   memory is the payload that travels with the lease.
+ *
+ * Note: This function declares the authority mode label. Actual claim/lease
+ * enforcement is delegated to the existing typed claim/lease owner in the
+ * control plane. This composition validates that the memory record itself
+ * meets the minimum bar for the declared mode (e.g., handoff-availability
+ * records for soft_claim/hard_lease).
  */
 export function composeHandoffWithMode(
   input: HandoffCompositionInput,
 ): HandoffCompositionResult {
   const { mode, memory, target_agent_id } = input;
 
+  // Validate that the memory record meets the minimum bar for the mode
+  if (mode === "soft_claim" || mode === "hard_lease") {
+    if (memory.availability !== "handoff") {
+      throw new EffectRuntimeRequestError(
+        `memory ${memory.memory_id} has availability "${memory.availability}" but mode "${mode}" requires "handoff"`,
+      );
+    }
+    if (!memory.public_safe) {
+      throw new EffectRuntimeRequestError(
+        `memory ${memory.memory_id} is not certified public_safe; cannot compose with mode "${mode}"`,
+      );
+    }
+  }
+
   const resumePacket = resumeFromMemory({
     memory,
     target_agent_id,

loopx/control_plane/memory/memory_projection.ts

@@ -39,21 +39,24 @@ export interface MemoryIndexProjection extends JsonObject {
 // Projection builder
 // ---------------------------------------------------------------------------
 
-export function buildMemoryIndexProjection(
+export async function buildMemoryIndexProjection(
   store: MemoryStore,
-): MemoryIndexProjection {
-  const index = store.getIndex();
-  const all = store.readAll();
+): Promise<MemoryIndexProjection> {
+  const index = await store.getIndex();
+  const all = await store.readAll();
   const byId = new Map(all.map((r) => [r.memory_id, r]));
 
   const activeCount = index.active_memories.length;
   const handoffCount = index.handoff_available.length;
 
+  // Only include public_safe records in the projection summary
+  const publicSafeRecords = all.filter((r) => r.public_safe);
+
   let latestMemoryAt: string | null = null;
   let latestSummary: string | null = null;
 
-  if (all.length > 0) {
-    const sorted = [...all].sort((a, b) => {
+  if (publicSafeRecords.length > 0) {
+    const sorted = [...publicSafeRecords].sort((a, b) => {
       if (a.created_at > b.created_at) return -1;
       if (a.created_at < b.created_at) return 1;
       return 0;
@@ -65,7 +68,7 @@ export function buildMemoryIndexProjection(
 
   const handoffRows: MemoryHandoffRow[] = index.handoff_available
     .map((id) => byId.get(id))
-    .filter((r): r is MemoryRecord => r !== undefined)
+    .filter((r): r is MemoryRecord => r !== undefined && r.public_safe)
     .map((r) => ({
       memory_id: r.memory_id,
       summary: r.summary,
@@ -100,15 +103,18 @@ export interface MemoryFirstScreen {
   readonly latest_summary: string | null;
 }
 
-export function buildMemoryFirstScreen(
+export async function buildMemoryFirstScreen(
   store: MemoryStore,
-): MemoryFirstScreen {
-  const index = store.getIndex();
-  const all = store.readAll();
+): Promise<MemoryFirstScreen> {
+  const index = await store.getIndex();
+  const all = await store.readAll();
+
+  // Only include public_safe records in the first-screen summary
+  const publicSafeRecords = all.filter((r) => r.public_safe);
 
   let latestSummary: string | null = null;
-  if (all.length > 0) {
-    const sorted = [...all].sort((a, b) => {
+  if (publicSafeRecords.length > 0) {
+    const sorted = [...publicSafeRecords].sort((a, b) => {
       if (a.created_at > b.created_at) return -1;
       if (a.created_at < b.created_at) return 1;
       return 0;

loopx/control_plane/memory/memory_record.ts

@@ -343,6 +343,8 @@ export interface CreateMemoryRecordInput {
   availability?: MemoryAvailability;
   source_run_id?: string | null;
   created_at?: string;
+  public_safe?: boolean;
+  redaction_verified?: boolean;
 }
 
 export function createMemoryRecord(input: CreateMemoryRecordInput): MemoryRecord {
@@ -366,8 +368,8 @@ export function createMemoryRecord(input: CreateMemoryRecordInput): MemoryRecord
     superseded_by: null,
     status: MEMORY_STATUS_ACTIVE,
     availability: input.availability ?? AVAILABILITY_SESSION,
-    public_safe: true,
-    redaction_verified: true,
+    public_safe: input.public_safe ?? false,
+    redaction_verified: input.redaction_verified ?? false,
     created_at: now,
     source_run_id: input.source_run_id ?? null,
   });

loopx/control_plane/memory/memory_resume.ts

@@ -120,17 +120,52 @@ export interface AdoptMemoryResult {
   readonly superseded_memory_id: string | null;
 }
 
-export function adoptMemory(
+export interface AdoptMemoryArgs {
+  readonly source_memory_id: string;
+  readonly successor_memory_id: string;
+}
+
+export async function adoptMemory(
   store: MemoryStore,
   packet: MemoryResumePacket,
-  args: { source_memory_id: string },
-): AdoptMemoryResult {
-  const { source_memory_id } = args;
+  args: AdoptMemoryArgs,
+): Promise<AdoptMemoryResult> {
+  const { source_memory_id, successor_memory_id } = args;
+
+  // Validate: successor must be distinct from source (no self-links)
+  if (source_memory_id === successor_memory_id) {
+    throw new EffectRuntimeRequestError(
+      `cannot adopt memory: successor_memory_id must differ from source_memory_id (${source_memory_id})`,
+    );
+  }
+
+  // Validate: successor must exist in the store
+  const successor = await store.readById(successor_memory_id);
+  if (!successor) {
+    throw new EffectRuntimeRequestError(
+      `cannot adopt memory: successor memory not found: ${successor_memory_id}`,
+    );
+  }
+
+  // Validate: successor must belong to the same goal
+  if (successor.goal_id !== packet.goal_id) {
+    throw new EffectRuntimeRequestError(
+      `cannot adopt memory: successor goal_id (${successor.goal_id}) does not match packet goal_id (${packet.goal_id})`,
+    );
+  }
+
+  // Validate: successor's supersedes must point back to source
+  if (successor.supersedes !== source_memory_id) {
+    throw new EffectRuntimeRequestError(
+      `cannot adopt memory: successor.supersedes (${successor.supersedes}) does not point to source_memory_id (${source_memory_id})`,
+    );
+  }
+
   // Mark the source memory as superseded
-  const index = store.getIndex();
+  const index = await store.getIndex();
   const source = index.supersede_chains[source_memory_id];
   if (source && source.superseded_by === null) {
-    store.updateSupersededBy(source_memory_id, packet.memory_id);
+    await store.updateSupersededBy(source_memory_id, successor_memory_id);
     return {
       adopted: true,
       resume_packet: packet,