Coverage for astrocyte/pipeline/provenance.py: 100%

1"""Where a memory came from, kept on its stored chunks. 

2 

3``retain(source=...)`` (a URL, a file path, an agent) used to reach only the 

4optional source store, so on the default path every recall hit's ``source`` 

5was ``None``. Retain now stamps it into each chunk's metadata under 

6:data:`SOURCE_KEY`, and every recall path reads it back with 

7:func:`stored_source`. 

8""" 

9 

10from __future__ import annotations 

11 

12from typing import Any 

13 

14#: Metadata key holding the retain request's ``source``. Underscore-prefixed: 

15#: written by the pipeline, not by callers (like ``_created_at``). 

16SOURCE_KEY = "_source" 

17 

18 

19def stored_source(metadata: dict[str, Any] | None) -> str | None: 

20 """The ``source`` a stored chunk was retained with, if any.""" 

21 value = (metadata or {}).get(SOURCE_KEY) 

22 return value if isinstance(value, str) and value else None