testing: replace the live-contract gate with timestampGranularity capability + peer-mention + readiness/drain fixes (comms-e5vm.10)

packages/core/ports.ts

@@ -10,7 +10,7 @@
  * ports.
  */
 
-import { Data, type Effect, type Option, Schema, type Stream } from 'effect'
+import { Data, type Duration, type Effect, type Option, Schema, type Stream } from 'effect'
 
 import { messageOf } from './messageOf.ts'
 
@@ -453,11 +453,34 @@ export interface Directory {
   presence(identity: Identity): Effect.Effect<Presence, DirectoryError>
 }
 
+/**
+ * Static, substrate-derived properties of an adapter's message-ordering
+ * model that a consumer must adapt to. Not behaviour — these are facts about
+ * the substrate the ports can't make uniform, surfaced so the same code
+ * (tests AND production) reads them rather than branching on a substrate name.
+ * Deliberately minimal: one field per real consumer, never a junk drawer of
+ * substrate flags.
+ */
+export interface Capabilities {
+  /**
+   * The smallest real-time gap between two `post`s that the substrate will
+   * stamp with distinct `Timestamp`s — the resolution of the ordering model.
+   * Memory keys `ts` off a monotonic counter so any two posts differ
+   * (`Duration.zero`); Zulip stamps integer epoch seconds, so posts inside the
+   * same second collide (`Duration.seconds(1)`). The gap-replay watermark
+   * dedups on `ts`, so "`ts` is not a unique key below this resolution" is
+   * knowledge production consults, not only a test concern — a caller that
+   * needs two posts distinguishable by `ts` must space them by at least this.
+   */
+  readonly timestampGranularity: Duration.Duration
+}
+
 /**
  * Aggregate exposed by a driven adapter. Driving adapters depend on
  * this shape, never on substrate-specific extensions.
  */
 export interface AgentComms {
+  readonly capabilities: Capabilities
   readonly identity: IdentityPort
   readonly publisher: MessagePublisher
   readonly inbox: MessageInbox

packages/mcp/memory-substrate.ts

@@ -1,7 +1,7 @@
-import type { AgentComms } from '@commy/core/ports'
+import type { AgentComms, Capabilities } from '@commy/core/ports'
 import type { ZulipAdapter } from '@commy/zulip/adapter'
 import { decodeUserUploadPathSync } from '@commy/zulip/http'
-import { Effect } from 'effect'
+import { Duration, Effect } from 'effect'
 
 /**
  * The single seam where above-the-port tests touch a Zulip type or brand.
@@ -44,16 +44,25 @@ const inertExtras: SubstrateExtras = {
   close: async () => {},
 }
 
+/**
+ * Above-port tests don't exercise timestamp granularity, so a hand-rolled port
+ * fake need not declare it; the inert default stands in (a real adapter passed
+ * as `base` overrides it via the spread).
+ */
+const inertCapabilities: Capabilities = { timestampGranularity: Duration.zero }
+
 /**
  * Complete an {@link AgentComms} core (the in-memory adapter, or a hand-rolled
  * port fake) to the `ZulipAdapter` shape the `SubstrateAdapter` port expects.
  * Pass overrides for the Zulip-shaped members a given test asserts on (e.g. a
- * counting `close`); the rest stay inert.
+ * counting `close`); the rest stay inert. `capabilities` may be omitted from a
+ * hand-rolled `base` — the inert default fills it.
  */
 export const completeAsSubstrate = (
-  base: AgentComms,
+  base: Omit<AgentComms, 'capabilities'> & Partial<Pick<AgentComms, 'capabilities'>>,
   overrides: SubstrateExtrasOverrides = {},
 ): ZulipAdapter => ({
+  capabilities: inertCapabilities,
   ...base,
   ...inertExtras,
   ...overrides,

packages/memory/adapter.ts

@@ -62,6 +62,7 @@ import {
   Array as Arr,
   Clock,
   Data,
+  Duration,
   Effect,
   HashMap,
   HashSet,
@@ -120,6 +121,20 @@ export type MemoryAdapter = AgentComms & {
    * tests and for modelling humans in MCP plugin tests.
    */
   readonly seedHuman: (name: string) => Effect.Effect<Identity, ParseResult.ParseError>
+  /**
+   * Inject a message AUTHORED BY `peer` (sender ≠ the bound self) into the
+   * substrate, running the same fan-out as a real post. The contract's
+   * mention-floor tests use this to prove a peer's @-mention of self surfaces
+   * on self's `events()` — a shape the single-identity `publisher.post`
+   * (always authored as self) cannot express. Unlike `post`, requires no
+   * bound identity: the peer is the author, not the adapter's own binding.
+   */
+  readonly peerPost: (
+    peer: Identity,
+    channel: ChannelRef,
+    body: MessageBodyType,
+    opts?: PostOpts,
+  ) => Effect.Effect<MessageRef, UnknownChannel>
 }
 
 const inRange =
@@ -664,7 +679,41 @@ export const memoryAdapter = (config: MemoryAdapterConfig = {}): Effect.Effect<M
     const seedHuman = (name: string): Effect.Effect<Identity, ParseResult.ParseError> =>
       registerIdentity(name, 'human')
 
+    // Authored by `peer`, not the bound self: mirrors publisher.post's store +
+    // fan-out path but stamps `sender: peer` and skips requireBound. The
+    // monotonic `nextTs` counter already yields a distinct ts per message, so
+    // memory reports `Duration.zero` granularity below — no spacing needed.
+    const peerPost = (
+      peer: Identity,
+      channel: ChannelRef,
+      body: MessageBodyType,
+      opts?: PostOpts,
+    ): Effect.Effect<MessageRef, UnknownChannel> =>
+      resolveBucket(channel).pipe(
+        Effect.flatMap((bucket) =>
+          Effect.gen(function* () {
+            const id = String(yield* allocId(nextMessageId))
+            const ref = yield* buildRef(id, channel, opts?.thread)
+            const ts = yield* decodeTimestamp(yield* allocId(nextTs)).pipe(Effect.orDie)
+            const stored: StoredMessage = {
+              ref,
+              sender: peer,
+              body,
+              ts,
+              mentions: opts?.mentions === undefined ? [] : [...opts.mentions],
+            }
+            bucket.push(stored)
+            messagesById.set(id, stored)
+            yield* fanOutOnPost(stored)
+            return ref
+          }),
+        ),
+      )
+
     return {
+      // Memory's `ts` is a monotonic counter, so any two posts already differ —
+      // no real-time spacing is needed for distinct timestamps.
+      capabilities: { timestampGranularity: Duration.zero },
       identity,
       publisher,
       inbox,
@@ -673,5 +722,6 @@ export const memoryAdapter = (config: MemoryAdapterConfig = {}): Effect.Effect<M
       seedChannel,
       seedAgent,
       seedHuman,
+      peerPost,
     }
   })

packages/memory/contract.test.ts

@@ -11,6 +11,8 @@ runAgentCommsContract('memory adapter', async () => {
     seedChannel: (name) => adapter.seedChannel(name).pipe(Effect.orDie),
     seedAgent: (name) => adapter.seedAgent(name).pipe(Effect.orDie),
     newUnacquiredAdapter: () => memoryAdapter(),
+    peerPost: (peer, channel, body, opts) =>
+      adapter.peerPost(peer, channel, body, opts).pipe(Effect.asVoid, Effect.orDie),
     dispose: () => Effect.void,
   }
 })
@@ -28,6 +30,8 @@ runAgentCommsContract('memory adapter (allowlist mode for UnknownIdentity covera
     seedAgent: (name) => adapter.seedAgent(name).pipe(Effect.orDie),
     newUnacquiredAdapter: () => memoryAdapter({ acquirableNames: ['hermes-agent-cycle'] }),
     unacquirableName: 'no-such-bot',
+    peerPost: (peer, channel, body, opts) =>
+      adapter.peerPost(peer, channel, body, opts).pipe(Effect.asVoid, Effect.orDie),
     dispose: () => Effect.void,
   }
 })