Skip to content

API

The reactive-research package provides the reference executable implementation of Reactive Research.

API documentation below is generated from the package source and docstrings.

For command-line usage, use:

uvx reactive-research --help
uvx reactive-research COMMAND --help

Models

reactive_research.models

Core data models for Reactive Research.

Diagnostic dataclass

One extraction or validation finding.

Source code in src/reactive_research/models.py 
64
65
66
67
68
69
70
71
@dataclass(frozen=True)
class Diagnostic:
    """One extraction or validation finding."""

    code: str
    message: str
    severity: DiagnosticSeverity
    source: SourceLocation | None = None

DiagnosticSeverity

Bases: StrEnum

Severity of one Reactive Research diagnostic.

Source code in src/reactive_research/models.py 
19
20
21
22
23
class DiagnosticSeverity(StrEnum):
    """Severity of one Reactive Research diagnostic."""

    ERROR = "error"
    WARNING = "warning"

ExtractionResult dataclass

Complete result of extracting one repository.

Source code in src/reactive_research/models.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@dataclass(frozen=True)
class ExtractionResult:
    """Complete result of extracting one repository."""

    repository: RepositoryContext
    declarations: tuple[ResearchDeclaration, ...] = field(default_factory=tuple)
    diagnostics: tuple[Diagnostic, ...] = field(default_factory=tuple)

    @property
    def defines(self) -> tuple[ResearchDeclaration, ...]:
        """Return all DEFINES declarations."""
        return tuple(
            declaration
            for declaration in self.declarations
            if declaration.relation is ResearchRelation.DEFINES
        )

    @property
    def implements(self) -> tuple[ResearchDeclaration, ...]:
        """Return all IMPLEMENTS declarations."""
        return tuple(
            declaration
            for declaration in self.declarations
            if declaration.relation is ResearchRelation.IMPLEMENTS
        )

    @property
    def valid(self) -> bool:
        """Return whether extraction produced no error diagnostics."""
        return not any(
            diagnostic.severity is DiagnosticSeverity.ERROR
            for diagnostic in self.diagnostics
        )

defines property 

defines: tuple[ResearchDeclaration, ...]

Return all DEFINES declarations.

implements property 

implements: tuple[ResearchDeclaration, ...]

Return all IMPLEMENTS declarations.

valid property 

valid: bool

Return whether extraction produced no error diagnostics.

RepositoryContext dataclass

Identity and revision of the repository being examined.

Source code in src/reactive_research/models.py 
54
55
56
57
58
59
60
61
@dataclass(frozen=True)
class RepositoryContext:
    """Identity and revision of the repository being examined."""

    root: Path
    organization: str | None
    name: str
    revision: str | None

ResearchDeclaration dataclass

One typed relationship to an addressable research object.

Source code in src/reactive_research/models.py 
45
46
47
48
49
50
51
@dataclass(frozen=True)
class ResearchDeclaration:
    """One typed relationship to an addressable research object."""

    relation: ResearchRelation
    identifier: ResearchIdentifier
    source: SourceLocation

ResearchIdentifier dataclass

Stable identifier for one addressable research object.

Source code in src/reactive_research/models.py 
26
27
28
29
30
31
32
33
34
@dataclass(frozen=True)
class ResearchIdentifier:
    """Stable identifier for one addressable research object."""

    value: str

    def __str__(self) -> str:
        """Return the canonical identifier string."""
        return self.value

__str__ 

__str__() -> str

Return the canonical identifier string.

Source code in src/reactive_research/models.py 
32
33
34
def __str__(self) -> str:
    """Return the canonical identifier string."""
    return self.value

ResearchRelation

Bases: StrEnum

Supported relationships between repositories and research objects.

Source code in src/reactive_research/models.py 
12
13
14
15
16
class ResearchRelation(StrEnum):
    """Supported relationships between repositories and research objects."""

    DEFINES = "DEFINES"
    IMPLEMENTS = "IMPLEMENTS"

SourceLocation dataclass

Location of one declaration in repository source.

Source code in src/reactive_research/models.py 
37
38
39
40
41
42
@dataclass(frozen=True)
class SourceLocation:
    """Location of one declaration in repository source."""

    path: Path
    line: int

Identifiers

reactive_research.identifiers

Research-object identifier parsing and validation.

Examples: SE-210.Definition.2.4 SE-210.Definition.4.3 SE-210.Theorem.5.2 SE-210.Proposition.4.14 SE-210.Example.4.11 SE-210.Remark.4.16c

InvalidResearchIdentifierError

Bases: ValueError

Raised when a research-object identifier is malformed.

Source code in src/reactive_research/identifiers.py 
26
27
class InvalidResearchIdentifierError(ValueError):
    """Raised when a research-object identifier is malformed."""

parse_research_identifier 

parse_research_identifier(value: str) -> ResearchIdentifier

Parse and validate a research-object identifier.

Source code in src/reactive_research/identifiers.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
def parse_research_identifier(value: str) -> ResearchIdentifier:
    """Parse and validate a research-object identifier."""
    normalized = value.strip()

    if not normalized:
        raise InvalidResearchIdentifierError("Research identifier must not be empty.")

    if not _IDENTIFIER_PATTERN.fullmatch(normalized):
        raise InvalidResearchIdentifierError(
            f"Invalid research identifier: {normalized!r}"
        )

    return ResearchIdentifier(normalized)

Repository Discovery

reactive_research.repository

Repository discovery for Reactive Research.

discover_repository_context 

discover_repository_context(
    path: Path,
) -> RepositoryContext

Discover repository identity and revision.

Source code in src/reactive_research/repository.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def discover_repository_context(path: Path) -> RepositoryContext:
    """Discover repository identity and revision."""
    root = path.resolve()

    organization: str | None = None
    name = root.name

    manifest_path = _find_manifest(root)
    if manifest_path is not None:
        manifest = _load_toml(manifest_path)

        repository = manifest.get("repository")
        if isinstance(repository, dict):
            repository_name = repository.get("name")
            repository_organization = repository.get("organization")

            if isinstance(repository_name, str) and repository_name:
                name = repository_name

            if isinstance(repository_organization, str) and repository_organization:
                organization = repository_organization

        # Compatibility with older SE manifests.
        repo = manifest.get("repo")
        if isinstance(repo, dict):
            repository_name = repo.get("name")
            if isinstance(repository_name, str) and repository_name:
                name = repository_name

    return RepositoryContext(
        root=root,
        organization=organization,
        name=name,
        revision=_git_revision(root),
    )

Extraction

reactive_research.extract

Extract Reactive Research declarations.

Reactive Research annotations may be provided in any source file with an eligible suffix.

extract_repository 

extract_repository(path: Path) -> ExtractionResult

Extract Reactive Research declarations from one repository.

Source code in src/reactive_research/extract.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def extract_repository(path: Path) -> ExtractionResult:
    """Extract Reactive Research declarations from one repository."""
    repository = discover_repository_context(path)

    declarations: list[ResearchDeclaration] = []
    diagnostics: list[Diagnostic] = []

    for source_path in _source_files(repository.root):
        source_declarations, source_diagnostics = _extract_file(
            source_path,
            root=repository.root,
        )
        declarations.extend(source_declarations)
        diagnostics.extend(source_diagnostics)

    diagnostics.extend(_duplicate_definition_diagnostics(declarations))

    return ExtractionResult(
        repository=repository,
        declarations=tuple(declarations),
        diagnostics=tuple(diagnostics),
    )

extract_research 

extract_research(
    *, path: Path, check: bool = False
) -> dict[str, Any]

Extract declarations and return the normalized CLI representation.

Source code in src/reactive_research/extract.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def extract_research(
    *,
    path: Path,
    check: bool = False,
) -> dict[str, Any]:
    """Extract declarations and return the normalized CLI representation."""
    result = extract_repository(path)
    document = declaration_document(result)

    return {
        "command": "extract",
        "check": check,
        **document,
    }

Validation

reactive_research.validate

Validate repository-local Reactive Research declarations.

validate_extraction 

validate_extraction(
    extraction: ExtractionResult, *, strict: bool = False
) -> tuple[Diagnostic, ...]

Validate one previously extracted repository result.

Source code in src/reactive_research/validate.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def validate_extraction(
    extraction: ExtractionResult,
    *,
    strict: bool = False,
) -> tuple[Diagnostic, ...]:
    """Validate one previously extracted repository result."""
    del strict

    diagnostics = list(extraction.diagnostics)

    repository = extraction.repository

    if not repository.name.strip():
        diagnostics.append(
            Diagnostic(
                code="RR.MISSING_REPOSITORY_NAME",
                message="Repository name must not be empty.",
                severity=DiagnosticSeverity.ERROR,
            )
        )

    if extraction.declarations and repository.revision is None:
        diagnostics.append(
            Diagnostic(
                code="RR.MISSING_REVISION",
                message=(
                    "Repository contains Reactive Research declarations "
                    "but no Git revision could be determined."
                ),
                severity=DiagnosticSeverity.WARNING,
            )
        )

    for declaration in extraction.declarations:
        if declaration.source.path.is_absolute():
            diagnostics.append(
                Diagnostic(
                    code="RR.ABSOLUTE_SOURCE_PATH",
                    message=("Declaration source paths must be repository-relative."),
                    severity=DiagnosticSeverity.ERROR,
                    source=declaration.source,
                )
            )

        if declaration.source.line < 1:
            diagnostics.append(
                Diagnostic(
                    code="RR.INVALID_SOURCE_LINE",
                    message=("Declaration source line must be greater than zero."),
                    severity=DiagnosticSeverity.ERROR,
                    source=declaration.source,
                )
            )

        try:
            parse_research_identifier(str(declaration.identifier))
        except InvalidResearchIdentifierError as error:
            diagnostics.append(
                Diagnostic(
                    code="RR.INVALID_IDENTIFIER",
                    message=str(error),
                    severity=DiagnosticSeverity.ERROR,
                    source=declaration.source,
                )
            )

    return tuple(diagnostics)

validate_research 

validate_research(
    *, path: Path, strict: bool = False
) -> dict[str, Any]

Validate repository-local Reactive Research declarations.

Source code in src/reactive_research/validate.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def validate_research(
    *,
    path: Path,
    strict: bool = False,
) -> dict[str, Any]:
    """Validate repository-local Reactive Research declarations."""
    extraction = extract_repository(path)
    diagnostics = validate_extraction(
        extraction,
        strict=strict,
    )

    document = declaration_document(extraction)

    valid = _diagnostics_are_valid(
        diagnostics,
        strict=strict,
    )

    return {
        "command": "validate",
        "path": str(extraction.repository.root),
        "repository": extraction.repository.name,
        "organization": extraction.repository.organization,
        "revision": extraction.repository.revision,
        "strict": strict,
        "valid": valid,
        "declaration_count": len(extraction.declarations),
        "declarations": document["declarations"],
        "diagnostics": [diagnostic_record(diagnostic) for diagnostic in diagnostics],
    }

Declaration Export

reactive_research.declarations

Normalized declaration documents for Reactive Research.

declaration_document 

declaration_document(
    result: ExtractionResult,
    *,
    extractor_version: str | None = None,
) -> dict[str, Any]

Return a normalized declaration document.

Source code in src/reactive_research/declarations.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def declaration_document(
    result: ExtractionResult,
    *,
    extractor_version: str | None = None,
) -> dict[str, Any]:
    """Return a normalized declaration document."""
    current_version = extractor_version or package_version()
    repository = result.repository

    return {
        "format": _DECLARATION_FORMAT,
        "format_version": _DECLARATION_FORMAT_VERSION,
        "extractor_version": current_version,
        "path": str(repository.root),
        "repository": {
            "organization": repository.organization,
            "name": repository.name,
            "revision": repository.revision,
        },
        "valid": result.valid,
        "declarations": [
            declaration_record(
                declaration,
                organization=repository.organization,
                repository=repository.name,
                revision=repository.revision,
                extractor_version=current_version,
            )
            for declaration in result.declarations
        ],
        "diagnostics": [
            diagnostic_record(diagnostic) for diagnostic in result.diagnostics
        ],
    }

declaration_record 

declaration_record(
    declaration: ResearchDeclaration,
    *,
    organization: str | None,
    repository: str,
    revision: str | None,
    extractor_version: str,
) -> dict[str, object]

Return one normalized research declaration.

Source code in src/reactive_research/declarations.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def declaration_record(
    declaration: ResearchDeclaration,
    *,
    organization: str | None,
    repository: str,
    revision: str | None,
    extractor_version: str,
) -> dict[str, object]:
    """Return one normalized research declaration."""
    source_path = declaration.source.path

    return {
        "identifier": str(declaration.identifier),
        "relation": declaration.relation.value,
        "organization": organization,
        "repository": repository,
        "revision": revision,
        "source_path": source_path.as_posix(),
        "source_line": declaration.source.line,
        "source_kind": _source_kind(source_path),
        "extractor_version": extractor_version,
    }

diagnostic_record 

diagnostic_record(
    diagnostic: Diagnostic,
) -> dict[str, object]

Return one normalized diagnostic.

Source code in src/reactive_research/declarations.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def diagnostic_record(
    diagnostic: Diagnostic,
) -> dict[str, object]:
    """Return one normalized diagnostic."""
    source_path: str | None = None
    source_line: int | None = None

    if diagnostic.source is not None:
        source_path = diagnostic.source.path.as_posix()
        source_line = diagnostic.source.line

    return {
        "code": diagnostic.code,
        "severity": diagnostic.severity.value,
        "message": diagnostic.message,
        "source_path": source_path,
        "source_line": source_line,
    }

export_declarations 

export_declarations(path: Path) -> dict[str, Any]

Extract and export normalized declarations for one repository.

Source code in src/reactive_research/declarations.py 
21
22
23
24
25
26
27
28
def export_declarations(path: Path) -> dict[str, Any]:
    """Extract and export normalized declarations for one repository."""
    # WHY: Local import avoids a module cycle because extract.py uses
    # declaration_document() for its machine-readable result.
    from reactive_research.extract import extract_repository

    result = extract_repository(path)
    return declaration_document(result)

is_declaration_document 

is_declaration_document(value: object) -> bool

Return whether a value has the normalized declaration-document shape.

Source code in src/reactive_research/declarations.py 
111
112
113
114
115
116
117
118
119
120
def is_declaration_document(value: object) -> bool:
    """Return whether a value has the normalized declaration-document shape."""
    if not isinstance(value, dict):
        return False

    return (
        value.get("format") == _DECLARATION_FORMAT
        and value.get("format_version") == _DECLARATION_FORMAT_VERSION
        and isinstance(value.get("declarations"), list)
    )

Resolution

reactive_research.resolve

Resolve Reactive Research identifiers.

RegistryLoadError

Bases: ValueError

Raised when declaration registry input cannot be loaded.

Source code in src/reactive_research/resolve.py 
24
25
class RegistryLoadError(ValueError):
    """Raised when declaration registry input cannot be loaded."""

resolve_research 

resolve_research(
    *,
    path: Path,
    identifier: str | None,
    resolve_all: bool,
    registry: str | None,
    snapshot: str | None,
) -> dict[str, Any]

Resolve research-object references.

Source code in src/reactive_research/resolve.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def resolve_research(
    *,
    path: Path,
    identifier: str | None,
    resolve_all: bool,
    registry: str | None,
    snapshot: str | None,
) -> dict[str, Any]:
    """Resolve research-object references."""
    current_document = export_declarations(path)

    if snapshot is not None:
        return {
            "command": "resolve",
            "path": str(path.resolve()),
            "identifier": identifier,
            "resolve_all": resolve_all,
            "registry": registry,
            "snapshot": snapshot,
            "resolved": False,
            "results": [],
            "diagnostics": [
                {
                    "code": "RR.SNAPSHOT_NOT_IMPLEMENTED",
                    "severity": "error",
                    "message": ("Snapshot-backed resolution is not implemented yet."),
                }
            ],
        }

    try:
        registry_documents = _load_registry_documents(
            registry,
            current_document=current_document,
        )
    except RegistryLoadError as error:
        return {
            "command": "resolve",
            "path": str(path.resolve()),
            "identifier": identifier,
            "resolve_all": resolve_all,
            "registry": registry,
            "snapshot": snapshot,
            "resolved": False,
            "results": [],
            "diagnostics": [
                {
                    "code": "RR.REGISTRY_LOAD_ERROR",
                    "severity": "error",
                    "message": str(error),
                }
            ],
        }

    targets, target_error = _resolution_targets(
        current_document=current_document,
        identifier=identifier,
        resolve_all=resolve_all,
    )

    if target_error is not None:
        return {
            "command": "resolve",
            "path": str(path.resolve()),
            "identifier": identifier,
            "resolve_all": resolve_all,
            "registry": registry,
            "snapshot": snapshot,
            "resolved": False,
            "results": [],
            "diagnostics": [target_error],
        }

    declarations = _registry_declarations(registry_documents)

    definitions = _definitions_by_identifier(declarations)

    results = [
        _resolve_identifier(
            target,
            definitions=definitions,
        )
        for target in targets
    ]

    resolved = all(result["status"] == "resolved" for result in results)

    return {
        "command": "resolve",
        "path": str(path.resolve()),
        "identifier": identifier,
        "resolve_all": resolve_all,
        "registry": registry,
        "snapshot": snapshot,
        "resolved": resolved,
        "results": results,
        "diagnostics": [],
    }

Graph

reactive_research.graph

Build and query Reactive Research graphs.

build_research_graph 

build_research_graph(
    *,
    path: Path,
    target: str | None,
    view: GraphView,
    registry: str | None,
    snapshot: str | None,
) -> dict[str, Any]

Build or inspect a typed Reactive Research graph.

Source code in src/reactive_research/graph.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def build_research_graph(
    *,
    path: Path,
    target: str | None,
    view: GraphView,
    registry: str | None,
    snapshot: str | None,
) -> dict[str, Any]:
    """Build or inspect a typed Reactive Research graph."""
    return {
        "command": "graph",
        "status": "scaffolded",
        "path": str(path.resolve()),
        "target": target,
        "view": view,
        "registry": registry,
        "snapshot": snapshot,
        "nodes": [],
        "edges": [],
    }

Impact Analysis

reactive_research.impact

Analyze downstream Reactive Research impact.

analyze_impact 

analyze_impact(
    *,
    path: Path,
    identifier: str,
    depth: ImpactDepth,
    registry: str | None,
    snapshot: str | None,
    from_snapshot: str | None,
    to_snapshot: str | None,
) -> dict[str, Any]

Identify downstream objects potentially affected by a change.

Source code in src/reactive_research/impact.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def analyze_impact(
    *,
    path: Path,
    identifier: str,
    depth: ImpactDepth,
    registry: str | None,
    snapshot: str | None,
    from_snapshot: str | None,
    to_snapshot: str | None,
) -> dict[str, Any]:
    """Identify downstream objects potentially affected by a change."""
    return {
        "command": "impact",
        "status": "scaffolded",
        "path": str(path.resolve()),
        "identifier": identifier,
        "depth": depth,
        "registry": registry,
        "snapshot": snapshot,
        "from_snapshot": from_snapshot,
        "to_snapshot": to_snapshot,
        "directly_affected": [],
        "transitively_affected": [],
        "revalidation_required": [],
    }

Snapshots

reactive_research.snapshot

Create and inspect Reactive Research graph snapshots.

snapshot_research_graph 

snapshot_research_graph(
    *,
    path: Path,
    registry: str | None,
    base_snapshot: str | None,
    show: str | None,
) -> dict[str, Any]

Create or inspect a reproducible graph snapshot.

Source code in src/reactive_research/snapshot.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def snapshot_research_graph(
    *,
    path: Path,
    registry: str | None,
    base_snapshot: str | None,
    show: str | None,
) -> dict[str, Any]:
    """Create or inspect a reproducible graph snapshot."""
    action = "show" if show is not None else "create"

    return {
        "command": "snapshot",
        "status": "scaffolded",
        "action": action,
        "path": str(path.resolve()),
        "registry": registry,
        "base_snapshot": base_snapshot,
        "snapshot_id": show,
        "repositories": [],
        "objects": [],
        "relationships": [],
    }

Inspection

reactive_research.inspect

Inspect Reactive Research objects and repositories.

inspect_research_object 

inspect_research_object(
    *,
    path: Path,
    target: str,
    registry: str | None,
    snapshot: str | None,
) -> dict[str, Any]

Explain one research object or repository.

Source code in src/reactive_research/inspect.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def inspect_research_object(
    *,
    path: Path,
    target: str,
    registry: str | None,
    snapshot: str | None,
) -> dict[str, Any]:
    """Explain one research object or repository."""
    return {
        "command": "inspect",
        "status": "scaffolded",
        "path": str(path.resolve()),
        "target": target,
        "registry": registry,
        "snapshot": snapshot,
        "defined_by": None,
        "implements": [],
        "implemented_by": [],
        "depends_on": [],
        "depended_on_by": [],
        "revalidation_required": False,
    }