Tag Derivation

akms.graph.tag_derivation

tag_derivation.py — Hybrid Tag Derivation Engine (§4 of schema spec).

Derives akms_tags for task JSONs when not explicitly set by the developer. Two strategies, run in sequence, results unioned:

1. Scope-based: Match task scope file paths against code-mirror node source_file fields and global node content_ref fields. Extract the owning node's tags.

2. Text-based: Concatenate title + objective + implementation_steps, then match against node titles (whole-word) and node tags (substring, min min_tag_length chars).

No LLM calls. Pure deterministic string matching.

derive_tags

derive_tags(
    G: DiGraph,
    task: dict,
    config: PropagationConfig | None = None,
) -> list[str]

Derive AKMS tags for a task using hybrid (scope + text) derivation.

If the task already has non-empty akms_tags, returns them unchanged (explicit tags from developer take precedence).

Parameters:

Name	Type	Description	Default
G	DiGraph	Compiled knowledge graph.	required
task	dict	Task JSON dict. Expected keys: - akms_tags (list[str], may be empty) - scope (list[str], file paths) - title (str) - objective (str) - implementation_steps (list[str])	required
config	PropagationConfig | None	PropagationConfig (uses tag_derivation section).	None

Returns:

Type	Description
list[str]	Sorted list of derived tags.

Source code in packages/akms/src/akms/graph/tag_derivation.py

@traced("akms.derive_tags")
def derive_tags(
    G: nx.DiGraph,
    task: dict,
    config: PropagationConfig | None = None,
) -> list[str]:
    """Derive AKMS tags for a task using hybrid (scope + text) derivation.

    If the task already has non-empty ``akms_tags``, returns them unchanged
    (explicit tags from developer take precedence).

    Args:
        G: Compiled knowledge graph.
        task: Task JSON dict. Expected keys:
            - ``akms_tags`` (list[str], may be empty)
            - ``scope`` (list[str], file paths)
            - ``title`` (str)
            - ``objective`` (str)
            - ``implementation_steps`` (list[str])
        config: PropagationConfig (uses ``tag_derivation`` section).

    Returns:
        Sorted list of derived tags.
    """
    if config is None:
        config = PropagationConfig()

    td_config = config.tag_derivation

    # Check for explicit tags
    existing = task.get("akms_tags", [])
    if existing:
        logger.debug("Task already has explicit akms_tags: %s", existing)
        return sorted(existing)

    # Strategy 1: scope-based
    scope = task.get("scope", [])
    scope_tags = _derive_tags_from_scope(G, scope)

    # Strategy 2: text-based
    text_tags = _derive_tags_from_text(G, task, td_config)

    all_tags = scope_tags | text_tags

    result = sorted(all_tags)

    if td_config.log_derived_tags:
        logger.info(
            "derive_tags: scope=%d text=%d total=%d tags=%s",
            len(scope_tags),
            len(text_tags),
            len(result),
            result,
        )

    return result

fill_task_tags

fill_task_tags(
    G: DiGraph,
    tasks: list[dict],
    config: PropagationConfig | None = None,
) -> list[dict]

Derive and fill akms_tags for a list of tasks.

Modifies tasks in-place and returns them. Only fills tags for tasks where akms_tags is empty or missing.

Parameters:

Name	Type	Description	Default
G	DiGraph	Compiled knowledge graph.	required
tasks	list[dict]	List of task JSON dicts.	required
config	PropagationConfig | None	PropagationConfig.	None

Returns:

Type	Description
list[dict]	The same list of tasks (modified in-place).

Source code in packages/akms/src/akms/graph/tag_derivation.py

def fill_task_tags(
    G: nx.DiGraph,
    tasks: list[dict],
    config: PropagationConfig | None = None,
) -> list[dict]:
    """Derive and fill akms_tags for a list of tasks.

    Modifies tasks in-place and returns them. Only fills tags for tasks
    where ``akms_tags`` is empty or missing.

    Args:
        G: Compiled knowledge graph.
        tasks: List of task JSON dicts.
        config: PropagationConfig.

    Returns:
        The same list of tasks (modified in-place).
    """
    for task in tasks:
        derived = derive_tags(G, task, config)
        task["akms_tags"] = derived
    return tasks

derive_review_seeds

derive_review_seeds(
    G: DiGraph,
    files_modified: list[str],
    fallback_tags: list[str] | None = None,
) -> list[str]

Seed tags for reviewer loadouts based on phase-branch changes.

Walks files_modified (from git diff against the phase-parent branch) and collects tags from: - Code-mirror nodes whose source_file matches (intended primary entry point per FR-T03). - Concept nodes whose content_ref matches.

When files_modified is empty (e.g. first phase, or the reviewer runs before any file change lands), fall back to fallback_tags — typically the union of completed-task akms_tags for this phase.

The return value is a deterministically ordered list so downstream query_subgraph calls produce reproducible loadouts.

Source code in packages/akms/src/akms/graph/tag_derivation.py

def derive_review_seeds(
    G: nx.DiGraph,
    files_modified: list[str],
    fallback_tags: list[str] | None = None,
) -> list[str]:
    """Seed tags for reviewer loadouts based on phase-branch changes.

    Walks ``files_modified`` (from ``git diff`` against the phase-parent
    branch) and collects tags from:
      - Code-mirror nodes whose ``source_file`` matches (intended primary
        entry point per FR-T03).
      - Concept nodes whose ``content_ref`` matches.

    When ``files_modified`` is empty (e.g. first phase, or the reviewer runs
    before any file change lands), fall back to ``fallback_tags`` — typically
    the union of completed-task ``akms_tags`` for this phase.

    The return value is a deterministically ordered list so downstream
    ``query_subgraph`` calls produce reproducible loadouts.
    """
    tags = _derive_tags_from_scope(G, list(files_modified or []))
    if not tags and fallback_tags:
        tags = {str(t) for t in fallback_tags if t}
    return sorted(tags)