`cosmotron_mcp.memory`¶

read_memory ¶

read_memory(task_description: str = '', tags: list[str] | None = None, top_k: int = 5) -> str

Retrieve cross-session memory records relevant to a task.

Pass the user's task verbatim as task_description to get the top-k most relevant AND mutually different active records (deterministic keyword scoring + diversity), grouped by type with failures/constraints first — this is the block to hand to @planner. Optional tags filter (e.g. ["shear", "masked"]). With no arguments, returns a one-line-per-record digest of the whole store instead.

Superseded records are never returned. Records live in repo-root MEMORY.md (override via the COSMOTRON_MEMORY_FILE env var).

Parameters:

Name	Type	Description	Default
`task_description`	`str`	User's task text verbatim; scored against stored records to select the top-k relevant AND mutually different ones. Empty string returns a one-line digest of the whole store instead.	`''`
`tags`	`list[str] \| None`	Optional keyword filter, e.g. `["shear", "masked"]`.	`None`
`top_k`	`int`	Number of records to return when task_description is given.	`5`

Returns:

Type	Description
`str`	A formatted memory block grouped by type (failures/constraints
`str`	first) when task_description is given, or a one-line-per-record
`str`	digest of the whole store otherwise.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def read_memory(
    task_description: str = "",
    tags: list[str] | None = None,
    top_k: int = 5,
) -> str:
    """Retrieve cross-session memory records relevant to a task.

    Pass the user's task verbatim as task_description to get the top-k most
    relevant AND mutually different active records (deterministic keyword
    scoring + diversity), grouped by type with failures/constraints first —
    this is the block to hand to @planner. Optional tags filter (e.g.
    ["shear", "masked"]). With no arguments, returns a one-line-per-record
    digest of the whole store instead.

    Superseded records are never returned. Records live in repo-root
    MEMORY.md (override via the COSMOTRON_MEMORY_FILE env var).

    Args:
        task_description: User's task text verbatim; scored against stored
            records to select the top-k relevant AND mutually different ones.
            Empty string returns a one-line digest of the whole store instead.
        tags: Optional keyword filter, e.g. ``["shear", "masked"]``.
        top_k: Number of records to return when task_description is given.

    Returns:
        A formatted memory block grouped by type (failures/constraints
        first) when task_description is given, or a one-line-per-record
        digest of the whole store otherwise.
    """
    return _read_memory(task_description=task_description, tags=tags, top_k=top_k)

append_memory ¶

append_memory(record_type: str, summary: str, detail: str, session_id: str, tags: list[str] | None = None, supersedes: str | None = None, force: bool = False) -> str

Append one memory record to repo-root MEMORY.md (human-gated; override path via the COSMOTRON_MEMORY_FILE env var).

record_type: one of mistake_fix, working_solution, methodology_note, pipeline_constraint. tags: short lowercase keywords aiding retrieval — suggested vocabulary: probe (clustering/shear), sky (fullsky/masked), tomography, and the key tools involved (e.g. pyccl, namaster, sacc, covariance, inference). Summaries must state the METHOD, never bake in pipeline-config numbers (nside=, lmax=, ...) — those live in session_context.json.

Returns "appended mem-NNN → " on success. On similarity to existing same-type records, returns a JSON {"status": "duplicate_candidates", ...} payload instead of writing — then either re-call with supersedes="mem-NNN" to REPLACE a stale record (it stays on disk but leaves retrieval), or with force=True to add as genuinely new. A config-polluted summary returns {"status": "config_polluted", ...}.

Parameters:

Name	Type	Description	Default
`record_type`	`str`	One of mistake_fix, working_solution, methodology_note, pipeline_constraint.	required
`summary`	`str`	One-sentence METHOD statement — never pipeline-config numbers (nside=, lmax=, ...); those live in session_context.json.	required
`detail`	`str`	Full explanation of the record.	required
`session_id`	`str`	Session this record originated from.	required
`tags`	`list[str] \| None`	Short lowercase keywords aiding retrieval — suggested vocabulary: probe (clustering/shear), sky (fullsky/masked), tomography, and the key tools involved (pyccl, namaster, sacc, covariance, inference).	`None`
`supersedes`	`str \| None`	`mem-NNN` of a stale record this one replaces (the old record stays on disk but is excluded from retrieval).	`None`
`force`	`bool`	Write as a new record even if it is similar to an existing same-type record.	`False`

Returns:

Type	Description
`str`	`"appended mem-NNN → <path>"` on success. A JSON
`str`	`{"status": "duplicate_candidates", ...}` payload instead of
`str`	writing when similar records exist (re-call with `supersedes=` or
`str`	`force=True`). `{"status": "config_polluted", ...}` if the
`str`	summary bakes in pipeline-config numbers.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def append_memory(
    record_type: str,
    summary: str,
    detail: str,
    session_id: str,
    tags: list[str] | None = None,
    supersedes: str | None = None,
    force: bool = False,
) -> str:
    """Append one memory record to repo-root MEMORY.md (human-gated; override
    path via the COSMOTRON_MEMORY_FILE env var).

    record_type: one of mistake_fix, working_solution, methodology_note,
    pipeline_constraint. tags: short lowercase keywords aiding retrieval —
    suggested vocabulary: probe (clustering/shear), sky (fullsky/masked),
    tomography, and the key tools involved (e.g. pyccl, namaster, sacc,
    covariance, inference). Summaries must state the METHOD, never bake in
    pipeline-config numbers (nside=, lmax=, ...) — those live in
    session_context.json.

    Returns "appended mem-NNN → <path>" on success. On similarity to existing
    same-type records, returns a JSON {"status": "duplicate_candidates", ...}
    payload instead of writing — then either re-call with supersedes="mem-NNN"
    to REPLACE a stale record (it stays on disk but leaves retrieval), or with
    force=True to add as genuinely new. A config-polluted summary returns
    {"status": "config_polluted", ...}.

    Args:
        record_type: One of mistake_fix, working_solution, methodology_note,
            pipeline_constraint.
        summary: One-sentence METHOD statement — never pipeline-config
            numbers (nside=, lmax=, ...); those live in session_context.json.
        detail: Full explanation of the record.
        session_id: Session this record originated from.
        tags: Short lowercase keywords aiding retrieval — suggested
            vocabulary: probe (clustering/shear), sky (fullsky/masked),
            tomography, and the key tools involved (pyccl, namaster, sacc,
            covariance, inference).
        supersedes: ``mem-NNN`` of a stale record this one replaces (the old
            record stays on disk but is excluded from retrieval).
        force: Write as a new record even if it is similar to an existing
            same-type record.

    Returns:
        ``"appended mem-NNN → <path>"`` on success. A JSON
        ``{"status": "duplicate_candidates", ...}`` payload instead of
        writing when similar records exist (re-call with ``supersedes=`` or
        ``force=True``). ``{"status": "config_polluted", ...}`` if the
        summary bakes in pipeline-config numbers.
    """
    return _append_memory(
        record_type, summary, detail, session_id,
        tags=tags, supersedes=supersedes, force=force,
    )

audit_memory ¶

audit_memory() -> str

Deterministic MEMORY.md health report (read-only JSON): per-type record counts, near-duplicate clusters among active same-type records, config-polluted summaries, and broken supersession links. Use it to propose consolidation to the user; acting on the report (editing the file, superseding records) stays human-gated.

Returns:

Type	Description
`str`	JSON health report: per-type record counts, near-duplicate clusters,
`str`	config-polluted summaries, and broken supersession links.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def audit_memory() -> str:
    """Deterministic MEMORY.md health report (read-only JSON): per-type record
    counts, near-duplicate clusters among active same-type records,
    config-polluted summaries, and broken supersession links. Use it to
    propose consolidation to the user; acting on the report (editing the file,
    superseding records) stays human-gated.

    Returns:
        JSON health report: per-type record counts, near-duplicate clusters,
        config-polluted summaries, and broken supersession links.
    """
    return _audit_memory()

cosmotron_mcp.memory¶

read_memory ¶

append_memory ¶

audit_memory ¶

`cosmotron_mcp.memory`¶