refactor(backend): shape W11 review nits ahead of CR

- records() takes ContentScope like store(); limit defaults to None and
  resolves to the repository protocol's DEFAULT_RECORD_LIMIT — one
  definition of the bound, no drifting literal in the service protocol
- read() docs state the whole-bytes return and its ~2x-blob peak as a
  deliberate v1 trade instead of claiming "streaming"; README matched
- StoredContentRecord.env is required — no default environment exists
  outside a scope

Author: lucas-xzp

src/backend/src/agentclaw/community/core/bot_config_manifest/README.md

@@ -195,7 +195,9 @@ store behind all three consumers of the requirement (audit, delivery,
   twice is two audit events, not one row to overwrite. Dedup lives in the
   blob layer by content address; repetition in the log is the fact.
 - **`read(digest)` is the one read path**, shared by delivery and audit:
-  streaming with the hash computed on the same pass, so the store returns
+  read in chunks with the hash computed on the same pass, returned whole
+  (peak ≈ 2× the blob at the §5 cap — a stated v1 trade; the consumer
+  materialises the full payload anyway), so the store returns
   bytes it can prove or fails — a re-delivery that "mostly" matches its
   address would defeat the receipt contract exactly where it matters. A
   missing address is terminal; this layer **never re-fetches** (§2.8's

src/backend/src/agentclaw/community/core/bot_config_manifest/content/models.py

@@ -61,7 +61,7 @@ class StoredContentRecord(BaseModel):
     """溯源行业务模型（对应 ac_manifest_content 表）。"""
 
     id: Optional[int] = Field(default=None, description="主键ID")
-    env: str = Field(default="dev", description="环境标识")
+    env: str = Field(..., description="环境标识（必填，无默认——scope 之外不存在「默认环境」）")
     entity_id: str = Field(..., description="实体ID（存储键，非公开字段）")
     bot_id: str = Field(..., description="Bot ID")
     digest: str = Field(..., description="内容地址 sha256:<hex64>")

src/backend/src/agentclaw/community/core/bot_config_manifest/content/service.py

@@ -234,11 +234,19 @@ def store(
     def read(self, digest: str) -> bytes:
         """The one read path, shared by delivery and audit (§2.8).
 
-        Streaming with the hash on the same pass: the store returns bytes it
-        can prove, or it raises — a re-delivery that "mostly" matches its
-        address would defeat the receipt contract exactly where it matters.
-        A missing address is terminal (``ContentMissingError``); this layer
-        never re-fetches.
+        The blob is read in chunks with the hash computed on the same pass,
+        and returned whole: the store hands back bytes it can prove, or it
+        raises — a re-delivery that "mostly" matches its address would
+        defeat the receipt contract exactly where it matters. A missing
+        address is terminal (``ContentMissingError``); this layer never
+        re-fetches.
+
+        Whole-bytes on purpose, at a stated cost: at the schema §5 cap
+        (100–200 MiB an entry) the peak is ~2× the blob (the chunk list and
+        the joined return). Apply is rare, and the read's consumer — W4's
+        delivery — materialises the full payload into the artifact anyway;
+        a chunk-wise contract is a decision for when a consumer exists that
+        wants one, not a default to speculate at now.
         """
         digest = _require_valid_digest(digest)
         blob = self._blob_path(digest)
@@ -259,15 +267,16 @@ def read(self, digest: str) -> bytes:
 
     def records(
         self,
+        scope: ContentScope,
         *,
-        env: str,
-        entity_id: str,
-        bot_id: str,
-        limit: int = DEFAULT_RECORD_LIMIT,
+        limit: Optional[int] = None,
     ) -> list[StoredContentRecord]:
         """The audit read: one bot's receipts, newest first."""
         return self._repository.records_for(
-            env=env, entity_id=entity_id, bot_id=bot_id, limit=limit
+            env=scope.env,
+            entity_id=scope.entity_id,
+            bot_id=scope.bot_id,
+            limit=DEFAULT_RECORD_LIMIT if limit is None else limit,
         )
 
     # --- the blob tree ----------------------------------------------------

src/backend/src/agentclaw/community/core/bot_config_manifest/content/service_protocol.py

@@ -80,11 +80,15 @@ def read(self, digest: str) -> bytes:
     @abstractmethod
     def records(
         self,
+        scope: ContentScope,
         *,
-        env: str,
-        entity_id: str,
-        bot_id: str,
-        limit: int = 50,  # == repository protocol's DEFAULT_RECORD_LIMIT; a literal keeps this protocol file domain-import-free at runtime
+        limit: Optional[int] = None,
     ) -> list[StoredContentRecord]:
-        """The audit read: one bot's receipts, newest first."""
+        """The audit read: one bot's receipts, newest first.
+
+        Same scope shape as ``store`` — the bot a store event was on behalf
+        of. ``limit=None`` means the repository protocol's
+        ``DEFAULT_RECORD_LIMIT``; that constant stays the one definition, so
+        this file carries no drifting copy of its value.
+        """
         ...

src/backend/tests/community/core/bot_config_manifest/content/test_content_store.py

@@ -106,9 +106,20 @@ def test_store_then_read_roundtrips_the_platform_copy(tmp_path):
     )
     assert stored.digest == BODY_SHA
     assert service.read(BODY_SHA) == BODY
-    # The audit read surfaces the same receipt.
-    assert [r.digest for r in service.records(
-        env=SCOPE.env, entity_id=SCOPE.entity_id, bot_id=SCOPE.bot_id)] == [BODY_SHA]
+    # The audit read surfaces the same receipt — scope-shaped, like store().
+    assert [r.digest for r in service.records(scope=SCOPE)] == [BODY_SHA]
+
+
+def test_records_limit_none_defaults_and_an_explicit_value_bounds(tmp_path):
+    # None means the repository protocol's DEFAULT_RECORD_LIMIT (the one
+    # definition of the bound — no drifting literal here); an explicit
+    # value bounds the audit read itself.
+    service, _ = _service(tmp_path)
+    service.store(_fetched(), scope=SCOPE, source_url="https://content.example/a.bin")
+    service.store(_fetched(url="https://mirror.example/a.bin"), scope=SCOPE,
+                  source_url="https://mirror.example/a.bin")
+    assert len(service.records(scope=SCOPE)) == 2
+    assert len(service.records(scope=SCOPE, limit=1)) == 1
 
 
 def test_a_receipt_that_disagrees_with_its_bytes_is_refused(tmp_path):