API reference

Every public function and the Theory class. The three functions whose name matches their module (diagram, dossier, simulate) are documented by their canonical path so the function, rather than the module, is shown.

theoryforge

theoryforge: systematic theory development.

A rigorous, reproducible workflow for building, developing and testing scientific theories. This is the feature-parity twin of the R package of the same name (https://pablobernabeu.github.io/theoryforge/r/). Every public function has an identically behaving counterpart there, pinned by the shared public specification (https://github.com/pablobernabeu/theoryforge/blob/main/API_SPEC.md), so the two implementations produce identical verdicts and byte-identical diagram intermediate representations. The only exceptions are the assistive helpers that depend on a network service or a user-supplied embedder (fetch_corpus, osf_push and embedding_redundancy) and the language-native diagram renderer (render_diagram), which are documented as such.

Theory

A theory as a versioned, machine-checkable object.

Wraps the parsed mapping (self.data); all accessors tolerate missing optional collections by treating them as empty.

Source code in src/theoryforge/core.py

class Theory:
    """A theory as a versioned, machine-checkable object.

    Wraps the parsed mapping (``self.data``); all accessors tolerate missing
    optional collections by treating them as empty.
    """

    def __init__(self, data: dict):
        if not isinstance(data, dict):
            raise TypeError("Theory data must be a mapping")
        self.data = data

    # -- convenience accessors -------------------------------------------------
    @property
    def id(self) -> str:
        return self.data.get("id", "")

    @property
    def maturity(self) -> str:
        return self.data.get("maturity", "")

    def _list(self, key: str) -> list:
        v = self.data.get(key)
        return v if isinstance(v, list) else []

    # -- validation ------------------------------------------------------------
    def validate(self, *, full: bool = False) -> bool:
        """Structural validation against the schema's required fields and enums.

        Returns True on success, raises ValueError listing every problem found.
        With ``full=True`` additionally checks referential integrity: that every
        id is unique within its collection and that every cross-reference
        (proposition endpoints, prediction derivations and diagnostics,
        assumption/evidence/test-outcome targets) points to a declared id. The
        ``full`` checks are deterministic.
        """
        errors: list[str] = []
        d = self.data
        for req in ("schema_version", "id", "title", "maturity"):
            if not _nonempty_str(d.get(req)):
                errors.append(f"missing/empty required field: {req}")
        if d.get("maturity") not in _MATURITY:
            errors.append(f"maturity must be one of {', '.join(sorted(_MATURITY))}")
        if "theory_form" in d and d["theory_form"] not in _FORM:
            errors.append(f"theory_form must be one of {', '.join(sorted(_FORM))}")
        for i, c in enumerate(self._list("constructs")):
            for req in ("id", "label", "definition"):
                if not _nonempty_str(c.get(req)):
                    errors.append(f"construct[{i}] missing/empty {req}")
        for i, p in enumerate(self._list("propositions")):
            for req in ("id", "from", "to", "relation"):
                if not _nonempty_str(p.get(req)):
                    errors.append(f"proposition[{i}] missing/empty {req}")
            if p.get("relation") not in _RELATION and _nonempty_str(p.get("relation")):
                errors.append(f"proposition[{i}] relation '{p.get('relation')}' not allowed")
        for i, p in enumerate(self._list("predictions")):
            for req in ("id", "statement", "type"):
                if not _nonempty_str(p.get(req)):
                    errors.append(f"prediction[{i}] missing/empty {req}")
            if p.get("type") not in _PRED_TYPE and _nonempty_str(p.get("type")):
                errors.append(f"prediction[{i}] type '{p.get('type')}' not allowed")

        if full:
            self._referential_errors(errors)

        if errors:
            raise ValueError("invalid theory object: " + "; ".join(errors))
        return True

    def _referential_errors(self, errors: list[str]) -> None:
        """Append referential-integrity problems (used by ``validate(full=True)``).

        Deterministic and mirrored byte-for-byte by the R implementation: the
        same checks in the same order with the same message text.
        """
        cons = self._list("constructs")
        props = self._list("propositions")
        preds = self._list("predictions")
        alts = self._list("alternatives")
        auxs = self._list("auxiliary_assumptions")
        construct_ids = {c.get("id") for c in cons if _nonempty_str(c.get("id"))}
        proposition_ids = {p.get("id") for p in props if _nonempty_str(p.get("id"))}
        prediction_ids = {p.get("id") for p in preds if _nonempty_str(p.get("id"))}
        alternative_ids = {a.get("id") for a in alts if _nonempty_str(a.get("id"))}

        def dups(items: list, kind: str) -> None:
            seen: set = set()
            for it in items:
                i = it.get("id")
                if _nonempty_str(i):
                    if i in seen:
                        errors.append(f"duplicate {kind} id: {i}")
                    seen.add(i)

        dups(cons, "construct")
        dups(props, "proposition")
        dups(preds, "prediction")
        dups(alts, "alternative")
        dups(auxs, "assumption")
        for i, p in enumerate(props):
            frm, to = p.get("from"), p.get("to")
            if _nonempty_str(frm) and frm not in construct_ids:
                errors.append(f"proposition[{i}] from '{frm}' is not a known construct")
            if _nonempty_str(to) and to not in construct_ids:
                errors.append(f"proposition[{i}] to '{to}' is not a known construct")
        for i, p in enumerate(preds):
            for dref in _as_list(p.get("derives_from")):
                if _nonempty_str(dref) and dref not in proposition_ids:
                    errors.append(f"prediction[{i}] derives_from '{dref}' is not a known proposition")
            for dv in _as_list(p.get("diagnostic_vs")):
                if _nonempty_str(dv) and dv not in alternative_ids:
                    errors.append(f"prediction[{i}] diagnostic_vs '{dv}' is not a known alternative")
        for i, a in enumerate(auxs):
            for pr in _as_list(a.get("protects")):
                if _nonempty_str(pr) and pr not in prediction_ids:
                    errors.append(f"assumption[{i}] protects '{pr}' is not a known prediction")
        for i, t in enumerate(self._list("test_outcomes")):
            pid = t.get("prediction_id")
            if _nonempty_str(pid) and pid not in prediction_ids:
                errors.append(f"test_outcome[{i}] prediction_id '{pid}' is not a known prediction")
        for i, e in enumerate(self._list("evidence")):
            s = e.get("supports")
            if _nonempty_str(s) and s not in prediction_ids:
                errors.append(f"evidence[{i}] supports '{s}' is not a known prediction")

    # -- serialisation ---------------------------------------------------------
    def write(self, path) -> None:
        path = Path(path)
        if path.suffix.lower() == ".json":
            path.write_text(json.dumps(self.data, indent=2, ensure_ascii=False), encoding="utf-8")
        else:
            path.write_text(
                yaml.safe_dump(self.data, sort_keys=False, allow_unicode=True),
                encoding="utf-8",
            )

    # -- builder (BUILDING mode) ----------------------------------------------
    def _provenance(self, action: str, detail: str) -> None:
        prov = self.data.setdefault("provenance", [])
        prov.append({"step": str(len(prov) + 1), "action": action, "detail": detail})

    def _coll(self, key: str) -> list:
        return self.data.setdefault(key, [])

    def add_construct(self, id, label, definition, measurement=None, boundary_conditions=None):
        c = {"id": id, "label": label, "definition": definition}
        if measurement is not None:
            c["measurement"] = list(measurement)
        if boundary_conditions is not None:
            c["boundary_conditions"] = list(boundary_conditions)
        self._coll("constructs").append(c)
        self._provenance("tf_add_construct", id)
        return self

    def add_proposition(self, id, frm, to, relation, mechanism=None):
        p = {"id": id, "from": frm, "to": to, "relation": relation}
        if mechanism is not None:
            p["mechanism"] = mechanism
        self._coll("propositions").append(p)
        self._provenance("tf_add_proposition", id)
        return self

    def add_prediction(self, id, statement, type, derives_from=None, diagnostic_vs=None):
        p = {"id": id, "statement": statement, "type": type}
        if derives_from is not None:
            p["derives_from"] = list(derives_from)
        if diagnostic_vs is not None:
            p["diagnostic_vs"] = list(diagnostic_vs)
        self._coll("predictions").append(p)
        self._provenance("tf_add_prediction", id)
        return self

    def add_alternative(self, id, label, key_constructs=None):
        a = {"id": id, "label": label}
        if key_constructs is not None:
            a["key_constructs"] = list(key_constructs)
        self._coll("alternatives").append(a)
        self._provenance("tf_add_alternative", id)
        return self

    def add_assumption(self, id, statement, added_for=None, protects=None):
        a = {"id": id, "statement": statement, "added_for": added_for}
        if protects is not None:
            a["protects"] = list(protects)
        self._coll("auxiliary_assumptions").append(a)
        self._provenance("tf_add_assumption", id)
        return self

    def set_formal_model(self, type, spec_ref=None):
        self.data["formal_model"] = {"type": type, "spec_ref": spec_ref}
        self._provenance("tf_set_formal_model", type)
        return self

    # -- mirrored public API --------------------------------------------------
    def check(self) -> dict:
        return _check(self.data)

    def report(self, format: str = "json") -> str:
        return _report(self.data, format=format)

    def redundancy_check(self) -> list[dict]:
        return _redundancy_check(self.data)

    def diagram(self, type: str = "nomological_net", engine: str = "graphviz") -> str:
        """Return the diagram IR for ``type`` (one of nomological_net, provenance,
        causal_dag, development_roadmap, pipeline, context, workflow, venn, rigour,
        severity)."""
        return _diagram(self.data, type=type, engine=engine)

    def render_diagram(self, type: str = "nomological_net"):
        """Render a digraph view via the optional ``graphviz`` library; see
        :func:`theoryforge.render.render_diagram`."""
        from .render import render_diagram as _render_diagram
        return _render_diagram(self.data, type=type)

    def severity(self) -> list[dict]:
        """Per-prediction risk and computed severity from the operationalised rubric."""
        return _severity(self.data)

    def appraise_amendment(self, prior) -> dict:
        """Progressive vs degenerating verdict for this theory relative to a prior version."""
        return _appraise_amendment(self.data, prior)

    def preregister(self, path=None) -> str:
        """Render a preregistration document (and write it if a path is given)."""
        return _preregister(self.data, path)

    def landscape(self, corpus, min_link: int = 2) -> dict:
        """Map this theory and its alternatives onto a literature corpus's themes."""
        return _landscape(self.data, corpus, min_link=min_link)

    def new_evidence_dois(self, candidate_dois: list) -> list:
        """DOIs in `candidate_dois` not already cited by this theory's evidence or alternatives."""
        return _new_evidence_dois(self.data, candidate_dois)

    def compile_sem(self) -> str:
        """Compile constructs and propositions to lavaan model syntax."""
        return _compile_sem(self.data)

    def dossier(self) -> str:
        """A reviewer-facing audit bundle of rigour report, severity, provenance, and preregistration."""
        return _dossier(self.data)

    def simulate(self, steps: int = 10, dt: float = 0.1, k: float = 1.0,
                 damping: float = 0.5, init: float = 1.0) -> dict:
        """Integrate the construct network as a linear dynamical system."""
        return _simulate(self.data, steps=steps, dt=dt, k=k, damping=damping, init=init)

    def embedding_redundancy(self, embedder, threshold=None) -> list[dict]:
        """An opt-in embedding-based redundancy screen; results depend on the supplied embedder."""
        return _embedding_redundancy(self.data, embedder, threshold=threshold)

    def render_report(self, path, title=None, render: bool = False, to: str = "html") -> str:
        """Write (and optionally render) a Quarto report of the audit dossier."""
        return _render_report(self.data, path, title=title, render=render, to=to)

    def osf_push(self, token=None, node=None, filename=None, dry_run: bool = True,
                 base_url: str | None = None) -> dict:
        """Deposit the dossier on OSF (dry-run by default). A live push needs a token and node."""
        kw = {} if base_url is None else {"base_url": base_url}
        return _osf_push(self.data, token=token, node=node, filename=filename, dry_run=dry_run, **kw)

    def __repr__(self) -> str:
        return f"Theory(id={self.id!r}, maturity={self.maturity!r})"

appraise_amendment(prior)

Progressive vs degenerating verdict for this theory relative to a prior version.

Source code in src/theoryforge/core.py

def appraise_amendment(self, prior) -> dict:
    """Progressive vs degenerating verdict for this theory relative to a prior version."""
    return _appraise_amendment(self.data, prior)

compile_sem()

Compile constructs and propositions to lavaan model syntax.

Source code in src/theoryforge/core.py

def compile_sem(self) -> str:
    """Compile constructs and propositions to lavaan model syntax."""
    return _compile_sem(self.data)

diagram(type='nomological_net', engine='graphviz')

Return the diagram IR for type (one of nomological_net, provenance, causal_dag, development_roadmap, pipeline, context, workflow, venn, rigour, severity).

Source code in src/theoryforge/core.py

def diagram(self, type: str = "nomological_net", engine: str = "graphviz") -> str:
    """Return the diagram IR for ``type`` (one of nomological_net, provenance,
    causal_dag, development_roadmap, pipeline, context, workflow, venn, rigour,
    severity)."""
    return _diagram(self.data, type=type, engine=engine)

dossier()

A reviewer-facing audit bundle of rigour report, severity, provenance, and preregistration.

Source code in src/theoryforge/core.py

def dossier(self) -> str:
    """A reviewer-facing audit bundle of rigour report, severity, provenance, and preregistration."""
    return _dossier(self.data)

embedding_redundancy(embedder, threshold=None)

An opt-in embedding-based redundancy screen; results depend on the supplied embedder.

Source code in src/theoryforge/core.py

def embedding_redundancy(self, embedder, threshold=None) -> list[dict]:
    """An opt-in embedding-based redundancy screen; results depend on the supplied embedder."""
    return _embedding_redundancy(self.data, embedder, threshold=threshold)

landscape(corpus, min_link=2)

Map this theory and its alternatives onto a literature corpus's themes.

Source code in src/theoryforge/core.py

def landscape(self, corpus, min_link: int = 2) -> dict:
    """Map this theory and its alternatives onto a literature corpus's themes."""
    return _landscape(self.data, corpus, min_link=min_link)

new_evidence_dois(candidate_dois)

DOIs in candidate_dois not already cited by this theory's evidence or alternatives.

Source code in src/theoryforge/core.py

def new_evidence_dois(self, candidate_dois: list) -> list:
    """DOIs in `candidate_dois` not already cited by this theory's evidence or alternatives."""
    return _new_evidence_dois(self.data, candidate_dois)

osf_push(token=None, node=None, filename=None, dry_run=True, base_url=None)

Deposit the dossier on OSF (dry-run by default). A live push needs a token and node.

Source code in src/theoryforge/core.py

def osf_push(self, token=None, node=None, filename=None, dry_run: bool = True,
             base_url: str | None = None) -> dict:
    """Deposit the dossier on OSF (dry-run by default). A live push needs a token and node."""
    kw = {} if base_url is None else {"base_url": base_url}
    return _osf_push(self.data, token=token, node=node, filename=filename, dry_run=dry_run, **kw)

preregister(path=None)

Render a preregistration document (and write it if a path is given).

Source code in src/theoryforge/core.py

def preregister(self, path=None) -> str:
    """Render a preregistration document (and write it if a path is given)."""
    return _preregister(self.data, path)

render_diagram(type='nomological_net')

Render a digraph view via the optional graphviz library; see :func:theoryforge.render.render_diagram.

Source code in src/theoryforge/core.py

def render_diagram(self, type: str = "nomological_net"):
    """Render a digraph view via the optional ``graphviz`` library; see
    :func:`theoryforge.render.render_diagram`."""
    from .render import render_diagram as _render_diagram
    return _render_diagram(self.data, type=type)

render_report(path, title=None, render=False, to='html')

Write (and optionally render) a Quarto report of the audit dossier.

Source code in src/theoryforge/core.py

def render_report(self, path, title=None, render: bool = False, to: str = "html") -> str:
    """Write (and optionally render) a Quarto report of the audit dossier."""
    return _render_report(self.data, path, title=title, render=render, to=to)

severity()

Per-prediction risk and computed severity from the operationalised rubric.

Source code in src/theoryforge/core.py

def severity(self) -> list[dict]:
    """Per-prediction risk and computed severity from the operationalised rubric."""
    return _severity(self.data)

simulate(steps=10, dt=0.1, k=1.0, damping=0.5, init=1.0)

Integrate the construct network as a linear dynamical system.

Source code in src/theoryforge/core.py

def simulate(self, steps: int = 10, dt: float = 0.1, k: float = 1.0,
             damping: float = 0.5, init: float = 1.0) -> dict:
    """Integrate the construct network as a linear dynamical system."""
    return _simulate(self.data, steps=steps, dt=dt, k=k, damping=damping, init=init)

validate(*, full=False)

Structural validation against the schema's required fields and enums.

Returns True on success, raises ValueError listing every problem found. With full=True additionally checks referential integrity: that every id is unique within its collection and that every cross-reference (proposition endpoints, prediction derivations and diagnostics, assumption/evidence/test-outcome targets) points to a declared id. The full checks are deterministic.

Source code in src/theoryforge/core.py

def validate(self, *, full: bool = False) -> bool:
    """Structural validation against the schema's required fields and enums.

    Returns True on success, raises ValueError listing every problem found.
    With ``full=True`` additionally checks referential integrity: that every
    id is unique within its collection and that every cross-reference
    (proposition endpoints, prediction derivations and diagnostics,
    assumption/evidence/test-outcome targets) points to a declared id. The
    ``full`` checks are deterministic.
    """
    errors: list[str] = []
    d = self.data
    for req in ("schema_version", "id", "title", "maturity"):
        if not _nonempty_str(d.get(req)):
            errors.append(f"missing/empty required field: {req}")
    if d.get("maturity") not in _MATURITY:
        errors.append(f"maturity must be one of {', '.join(sorted(_MATURITY))}")
    if "theory_form" in d and d["theory_form"] not in _FORM:
        errors.append(f"theory_form must be one of {', '.join(sorted(_FORM))}")
    for i, c in enumerate(self._list("constructs")):
        for req in ("id", "label", "definition"):
            if not _nonempty_str(c.get(req)):
                errors.append(f"construct[{i}] missing/empty {req}")
    for i, p in enumerate(self._list("propositions")):
        for req in ("id", "from", "to", "relation"):
            if not _nonempty_str(p.get(req)):
                errors.append(f"proposition[{i}] missing/empty {req}")
        if p.get("relation") not in _RELATION and _nonempty_str(p.get("relation")):
            errors.append(f"proposition[{i}] relation '{p.get('relation')}' not allowed")
    for i, p in enumerate(self._list("predictions")):
        for req in ("id", "statement", "type"):
            if not _nonempty_str(p.get(req)):
                errors.append(f"prediction[{i}] missing/empty {req}")
        if p.get("type") not in _PRED_TYPE and _nonempty_str(p.get("type")):
            errors.append(f"prediction[{i}] type '{p.get('type')}' not allowed")

    if full:
        self._referential_errors(errors)

    if errors:
        raise ValueError("invalid theory object: " + "; ".join(errors))
    return True

read(path)

Read a theory object from a YAML or JSON file.

Source code in src/theoryforge/core.py

def read(path) -> Theory:
    """Read a theory object from a YAML or JSON file."""
    path = Path(path)
    text = path.read_text(encoding="utf-8")
    data = json.loads(text) if path.suffix.lower() == ".json" else yaml.safe_load(text)
    if not isinstance(data, dict):
        raise ValueError("Theory data must be a mapping")
    return Theory(data)

write(theory, path)

Write a theory object to YAML or JSON (chosen by file extension).

Source code in src/theoryforge/core.py

def write(theory: Theory, path) -> None:
    """Write a theory object to YAML or JSON (chosen by file extension)."""
    theory.write(path)

new_theory(id, title, maturity='building', theory_form='network')

Start a new, empty theory object (BUILDING mode entry point).

Source code in src/theoryforge/core.py

def new_theory(id: str, title: str, maturity: str = "building", theory_form: str = "network") -> Theory:
    """Start a new, empty theory object (BUILDING mode entry point)."""
    t = Theory({
        "schema_version": "1.0",
        "id": id,
        "title": title,
        "maturity": maturity,
        "theory_form": theory_form,
    })
    t._provenance("tf_theory", id)
    return t

check(T)

Compute the full rigour report (dict) for a Theory or theory mapping.

Source code in src/theoryforge/rigor.py

def check(T) -> dict:
    """Compute the full rigour report (dict) for a Theory or theory mapping."""
    T = T.data if hasattr(T, "data") else T
    spec = _resources.checklist()
    thr = spec["thresholds"]
    results = _check_items(T, thr)

    items = []
    weighted = 0.0
    n_blockers_failed = 0
    for spec_item in spec["items"]:
        iid = spec_item["id"]
        status, score = results[iid]
        weighted += spec_item["weight"] * score
        if spec_item["severity_if_fail"] == "blocker" and status == "fail":
            n_blockers_failed += 1
        items.append({
            "id": iid,
            "status": status,
            "score": score,
            "weight": spec_item["weight"],
            "severity_if_fail": spec_item["severity_if_fail"],
            "citation": spec_item["citation"],
        })

    maturity = T.get("maturity", "")
    if maturity == "draft":
        gate = "advisory"
    else:
        gate = "blocked" if n_blockers_failed > 0 else "pass"

    return {
        "theory_id": T.get("id", ""),
        "schema_version": T.get("schema_version", ""),
        "maturity": maturity,
        "aggregate_score": rnd(weighted * 100, 1),
        "gate": gate,
        "n_blockers_failed": n_blockers_failed,
        "items": items,
    }

report(T, format='json')

Render the rigour report. format in {'json', 'html'}.

Source code in src/theoryforge/rigor.py

def report(T, format: str = "json") -> str:
    """Render the rigour report. format in {'json', 'html'}."""
    rep = check(T)  # check() unwraps Theory -> mapping
    if format == "json":
        return json.dumps(rep, indent=2, ensure_ascii=False)
    if format == "html":
        rows = "\n".join(
            f'    <tr><td>{it["id"]}</td><td>{it["status"]}</td>'
            f'<td>{it["score"]}</td><td>{it["citation"]}</td></tr>'
            for it in rep["items"]
        )
        return (
            f'<section class="theoryforge-report">\n'
            f'  <h2>Rigour report: {rep["theory_id"]}</h2>\n'
            f'  <p>Aggregate score: <strong>{rep["aggregate_score"]}</strong> &middot; '
            f'gate: <strong>{rep["gate"]}</strong></p>\n'
            f'  <table>\n    <tr><th>item</th><th>status</th><th>score</th><th>grounding</th></tr>\n'
            f'{rows}\n  </table>\n</section>\n'
        )
    raise ValueError(f"unknown report format: {format!r}")

redundancy_check(T)

Pairwise lexical similarity of construct definitions.

Returns one record per unordered construct pair, sorted by descending similarity then (a, b) ascending.

References

Le, H., Schmidt, F. L., Harter, J. K., & Lauver, K. J. (2010). The problem of empirical redundancy of constructs. Organizational Behavior and Human Decision Processes, 112(2), 112-125. https://doi.org/10.1016/j.obhdp.2010.02.003 Lawson, K. M., & Robins, R. W. (2021). Sibling constructs. Personality and Social Psychology Review, 25(4), 344-366. https://doi.org/10.1177/10888683211047101

Source code in src/theoryforge/redundancy.py

def redundancy_check(T: dict) -> list[dict]:
    """Pairwise lexical similarity of construct definitions.

    Returns one record per unordered construct pair, sorted by descending
    similarity then ``(a, b)`` ascending.

    References:
        Le, H., Schmidt, F. L., Harter, J. K., & Lauver, K. J. (2010). The
        problem of empirical redundancy of constructs. Organizational Behavior
        and Human Decision Processes, 112(2), 112-125.
        https://doi.org/10.1016/j.obhdp.2010.02.003
        Lawson, K. M., & Robins, R. W. (2021). Sibling constructs. Personality
        and Social Psychology Review, 25(4), 344-366.
        https://doi.org/10.1177/10888683211047101
    """
    T = T.data if hasattr(T, "data") else T
    cons = _list(T, "constructs")
    thr = _resources.checklist()["thresholds"]["redundancy_similarity_max"]
    toks = [(c.get("id", ""), tokens(c.get("definition", ""))) for c in cons]
    rows = []
    for i in range(len(toks)):
        for j in range(i + 1, len(toks)):
            sim = jaccard(toks[i][1], toks[j][1])
            rows.append({
                "a": toks[i][0],
                "b": toks[j][0],
                "similarity": sim,
                "flag": "review" if sim >= thr else "ok",
            })
    rows.sort(key=lambda r: (-r["similarity"], r["a"], r["b"]))
    return rows

tokens(s)

Tokenise a string into a set of content tokens.

Source code in src/theoryforge/redundancy.py

def tokens(s: str) -> set[str]:
    """Tokenise a string into a set of content tokens."""
    s = (s or "").lower()
    parts = _NON_ALNUM.sub(" ", s).split()
    return {t for t in parts if len(t) >= 3 and t not in STOPWORDS}

jaccard(a, b)

Source code in src/theoryforge/redundancy.py

def jaccard(a: set[str], b: set[str]) -> float:
    if not a and not b:
        return 0.0
    inter = len(a & b)
    union = len(a | b)
    return rnd(inter / union, 3)

severity(T)

Per-prediction risk and computed severity, in file order.

References

Mayo, D. G. (2018). Statistical inference as severe testing. Cambridge University Press. https://doi.org/10.1017/9781107286184 Meehl, P. E. (1990). Why summaries of research on psychological theories are often uninterpretable. Psychological Reports, 66, 195-244. https://doi.org/10.2466/pr0.1990.66.1.195

Source code in src/theoryforge/scoring.py

def severity(T) -> list[dict]:
    """Per-prediction risk and computed severity, in file order.

    References:
        Mayo, D. G. (2018). Statistical inference as severe testing. Cambridge
        University Press. https://doi.org/10.1017/9781107286184
        Meehl, P. E. (1990). Why summaries of research on psychological theories
        are often uninterpretable. Psychological Reports, 66, 195-244.
        https://doi.org/10.2466/pr0.1990.66.1.195
    """
    T = T.data if hasattr(T, "data") else T
    preds = _list(T, "predictions")
    alt_ids = {a.get("id") for a in _list(T, "alternatives")}
    out = []
    for p in preds:
        typ = p.get("type")
        base = BASE.get(typ, 0.0)
        discounted = base * (1 - CRUD) if typ == "directional" else base
        dv = _as_list(p.get("diagnostic_vs"))
        diag_bonus = 0.1 if dv and any(d in alt_ids for d in dv) else 0.0
        out.append({
            "prediction_id": p.get("id"),
            "type": typ,
            "risk_score": rnd(base, 3),
            "computed_severity": rnd(min(1.0, discounted + diag_bonus), 3),
        })
    return out

appraise_amendment(new, prior)

Appraise an amendment as progressive, degenerating, or neutral relative to a prior version.

References

Lakatos, I. (1970). Falsification and the methodology of scientific research programmes. In Criticism and the growth of knowledge (pp. 91-196). Cambridge University Press. https://doi.org/10.1017/cbo9781139171434.009 Meehl, P. E. (1990). Appraising and amending theories. Psychological Inquiry, 1(2), 108-141. https://doi.org/10.1207/s15327965pli0102_1

Source code in src/theoryforge/develop.py

def appraise_amendment(new, prior) -> dict:
    """Appraise an amendment as progressive, degenerating, or neutral relative to a prior version.

    References:
        Lakatos, I. (1970). Falsification and the methodology of scientific
        research programmes. In Criticism and the growth of knowledge
        (pp. 91-196). Cambridge University Press.
        https://doi.org/10.1017/cbo9781139171434.009
        Meehl, P. E. (1990). Appraising and amending theories. Psychological
        Inquiry, 1(2), 108-141. https://doi.org/10.1207/s15327965pli0102_1
    """
    new = new.data if hasattr(new, "data") else new
    prior = prior.data if hasattr(prior, "data") else prior

    prior_pred_ids = {p.get("id") for p in _list(prior, "predictions")}
    prior_aux_ids = {a.get("id") for a in _list(prior, "auxiliary_assumptions")}
    tos = _list(new, "test_outcomes")

    def passed(pid: str) -> bool:
        return any(t.get("prediction_id") == pid and t.get("passed") is True for t in tos)

    new_predictions = [p.get("id") for p in _list(new, "predictions") if p.get("id") not in prior_pred_ids]
    corroborated_new = [pid for pid in new_predictions if passed(pid)]

    ad_hoc = []
    for a in _list(new, "auxiliary_assumptions"):
        if a.get("id") in prior_aux_ids:
            continue
        if a.get("added_for") is None:
            continue
        protects = _as_list(a.get("protects"))
        if not any(t.get("prediction_id") in protects and t.get("passed") is True for t in tos):
            ad_hoc.append(a.get("id"))

    if len(corroborated_new) >= 1 and len(ad_hoc) == 0:
        verdict = "progressive"
    elif len(ad_hoc) >= 1 and len(corroborated_new) == 0:
        verdict = "degenerating"
    else:
        verdict = "neutral"

    return {
        "verdict": verdict,
        "new_predictions": sorted(new_predictions),
        "corroborated_new": sorted(corroborated_new),
        "ad_hoc_assumptions": sorted(ad_hoc),
    }

preregister(T, path=None)

Render a preregistration document, writing it to path when one is given.

Source code in src/theoryforge/prereg.py

def preregister(T, path=None) -> str:
    """Render a preregistration document, writing it to ``path`` when one is given."""
    data = T.data if hasattr(T, "data") else T
    rep = _check(data)
    deriv = next((it for it in rep["items"] if it["id"] == "derivation_chain"), None)
    verified = "yes" if deriv and deriv["status"] == "pass" else "no"

    lines = [
        f"# Preregistration: {data.get('title', '')}",
        "",
        f"- Theory ID: {data.get('id', '')}",
        f"- Schema version: {data.get('schema_version', '')}",
        f"- Maturity: {data.get('maturity', '')}",
        f"- Derivation chain verified: {verified}",
        "",
        "## Hypotheses",
    ]
    preds = _list(data, "predictions")
    if not preds:
        lines.append("_No predictions specified._")
    else:
        for i, p in enumerate(preds, start=1):
            df = _as_list(p.get("derives_from"))
            df_txt = ", ".join(df) if df else "—"
            lines.append(f"{i}. [{p.get('type')}] {p.get('statement')} (derives from: {df_txt})")

    lines += ["", "## Severity"]
    sev = _severity(data)
    if not sev:
        lines.append("_No predictions specified._")
    else:
        for s in sev:
            pid, cs, rk = s["prediction_id"], _fmt(s["computed_severity"]), _fmt(s["risk_score"])
            lines.append(f"- {pid}: severity {cs}, risk {rk}")

    text = "\n".join(lines) + "\n"
    if path is not None:
        Path(path).write_bytes(text.encode("utf-8"))
    return text

read_corpus(path)

Read a literature corpus from YAML or JSON.

Source code in src/theoryforge/lit.py

def read_corpus(path) -> dict:
    """Read a literature corpus from YAML or JSON."""
    path = Path(path)
    text = path.read_text(encoding="utf-8")
    data = json.loads(text) if path.suffix.lower() == ".json" else yaml.safe_load(text)
    if not isinstance(data, dict):
        raise ValueError("Corpus data must be a mapping")
    return data

litmap(corpus, min_link=DEFAULT_MIN_LINK)

Keyword co-occurrence, thematic components, and co-citation, all deterministic.

Source code in src/theoryforge/lit.py

def litmap(corpus, min_link: int = DEFAULT_MIN_LINK) -> dict:
    """Keyword co-occurrence, thematic components, and co-citation, all deterministic."""
    records = _records(corpus)
    all_keywords = sorted({k for r in records for k in _as_list(r.get("keywords")) if k})
    kw_edges = _edges(_pair_counts(records, "keywords"), min_link)
    cocit = _edges(_pair_counts(records, "references"), min_link)
    return {
        "n_records": len(records),
        "keywords": all_keywords,
        "keyword_cooccurrence": kw_edges,
        "themes": _components(kw_edges),
        "co_citation": cocit,
    }

landscape(theory, corpus, min_link=DEFAULT_MIN_LINK)

Map a theory and its registered alternatives onto the literature's themes.

Source code in src/theoryforge/lit.py

def landscape(theory, corpus, min_link: int = DEFAULT_MIN_LINK) -> dict:
    """Map a theory and its registered alternatives onto the literature's themes."""
    T = theory.data if hasattr(theory, "data") else theory
    lm = litmap(corpus, min_link)

    focal_src = " ".join(
        [T.get("title", "")] + [c.get("label", "") for c in (T.get("constructs") or [])]
    )
    focal_tokens = tokens(focal_src)
    alts = T.get("alternatives") or []

    themes_out = []
    under, crowded = [], []
    for th in lm["themes"]:
        th_tokens = tokens(" ".join(th["keywords"]))
        on = sorted(
            a.get("id") for a in alts
            if tokens(a.get("label", "") + " " + " ".join(_as_list(a.get("key_constructs")))) & th_tokens
        )
        focal_on = bool(focal_tokens & th_tokens)
        n = len(on) + (1 if focal_on else 0)
        status = "under_theorised" if n == 0 else ("crowded" if n >= 2 else "covered")
        themes_out.append({
            "id": th["id"], "keywords": th["keywords"],
            "alternatives": on, "focal": focal_on, "status": status,
        })
        if status == "under_theorised":
            under.append(th["id"])
        elif status == "crowded":
            crowded.append(th["id"])

    return {
        "theory_id": T.get("id", ""),
        "themes": themes_out,
        "under_theorised_fronts": under,
        "redundancy_risk": crowded,
    }

lit_diagram(obj, type='keyword_cooccurrence')

DOT for the literature layer. type in {keyword_cooccurrence, co_citation, theme_landscape}.

Source code in src/theoryforge/lit.py

def lit_diagram(obj: dict, type: str = "keyword_cooccurrence") -> str:
    """DOT for the literature layer. type in {keyword_cooccurrence, co_citation, theme_landscape}."""
    if type in ("keyword_cooccurrence", "co_citation"):
        return _undirected(type, obj.get(type, []))
    if type == "theme_landscape":
        return _theme_landscape(obj)
    raise ValueError(
        f"unknown lit diagram type {type!r}; expected one of "
        "('keyword_cooccurrence', 'co_citation', 'theme_landscape')"
    )

fetch_corpus(query, per_page=25, mailto=None)

Build a corpus from the OpenAlex API (network call).

This adapter is assistive. It depends on a live external service whose results change over time, so it sits outside the package's deterministic core.

Source code in src/theoryforge/lit.py

def fetch_corpus(query: str, per_page: int = 25, mailto: str | None = None) -> dict:
    """Build a corpus from the OpenAlex API (network call).

    This adapter is assistive. It depends on a live external service whose results
    change over time, so it sits outside the package's deterministic core.
    """
    import urllib.parse
    import urllib.request

    params = {"search": query, "per-page": str(per_page)}
    if mailto:
        params["mailto"] = mailto
    url = "https://api.openalex.org/works?" + urllib.parse.urlencode(params)
    with urllib.request.urlopen(url, timeout=30) as resp:  # noqa: S310 (documented external call)
        data = json.load(resp)

    records = []
    for w in data.get("results", []):
        kws = [k.get("display_name") for k in (w.get("keywords") or [])]
        if not kws:
            kws = [c.get("display_name") for c in (w.get("concepts") or [])[:5]]
        records.append({
            "id": w.get("id"),
            "title": w.get("title"),
            "year": w.get("publication_year"),
            "keywords": [k for k in kws if k],
            "references": w.get("referenced_works") or [],
        })
    return {"schema_version": "1.0", "id": f"openalex:{query}", "records": records}

new_evidence_dois(theory, candidate_dois)

DOIs in candidate_dois not already cited by the theory's evidence or alternatives.

Compares by normalised form (lowercased, with any doi.org/dx.doi.org URL prefix stripped), so a fresh literature search, for example via OpenAlex, Scopus, or any other source, can be checked against what the theory already engages with. Returns the qualifying DOIs in their original form, deduplicated and sorted by normalised form. Deterministic and takes no network dependency: the search itself is left to whichever literature tool the caller prefers.

Source code in src/theoryforge/lit.py

def new_evidence_dois(theory, candidate_dois: list) -> list:
    """DOIs in `candidate_dois` not already cited by the theory's evidence or alternatives.

    Compares by normalised form (lowercased, with any doi.org/dx.doi.org URL prefix
    stripped), so a fresh literature search, for example via OpenAlex, Scopus, or any
    other source, can be checked against what the theory already engages with. Returns
    the qualifying DOIs in their original form, deduplicated and sorted by normalised
    form. Deterministic and takes no network dependency: the search itself is left to
    whichever literature tool the caller prefers.
    """
    T = theory.data if hasattr(theory, "data") else theory
    known = set()
    for e in T.get("evidence") or []:
        doi = e.get("source_doi")
        if doi:
            known.add(_normalize_doi(doi))
    for a in T.get("alternatives") or []:
        doi = a.get("source_doi")
        if doi:
            known.add(_normalize_doi(doi))

    seen, out = set(), []
    for doi in candidate_dois or []:
        if not doi:
            continue
        norm = _normalize_doi(doi)
        if norm in known or norm in seen:
            continue
        seen.add(norm)
        out.append(doi)
    return sorted(out, key=_normalize_doi)

compile_sem(T)

Source code in src/theoryforge/sem.py

def compile_sem(T) -> str:
    T = T.data if hasattr(T, "data") else T
    out = [f"# lavaan model generated by theoryforge for {T.get('id', '')}", "# Measurement model"]
    for c in _list(T, "constructs"):
        meas = _as_list(c.get("measurement"))
        if meas:
            out.append(f"{c.get('id')} =~ " + " + ".join(_san(m) for m in meas))
    out.append("# Structural model")
    for p in _list(T, "propositions"):
        rel, frm, to = p.get("relation"), p.get("from"), p.get("to")
        if rel in _PATH:
            out.append(f"{to} ~ {frm}")
        elif rel == "associates":
            out.append(f"{frm} ~~ {to}")
        elif rel == "moderates":
            out.append(f"# moderation: {frm} moderates the path into {to} (specify interaction manually)")
    return "\n".join(out) + "\n"

embedding_redundancy(T, embedder, threshold=None)

Pairwise cosine similarity of embedded construct definitions.

embedder maps a definition string to a numeric vector. Returns one record per unordered construct pair, sorted by descending similarity then (a, b), with a review/ok flag.

Source code in src/theoryforge/embedding.py

def embedding_redundancy(T, embedder: Callable[[str], Sequence[float]],
                         threshold: float | None = None) -> list[dict]:
    """Pairwise cosine similarity of embedded construct definitions.

    `embedder` maps a definition string to a numeric vector. Returns one record per unordered
    construct pair, sorted by descending similarity then (a, b), with a `review`/`ok` flag.
    """
    T = T.data if hasattr(T, "data") else T
    if threshold is None:
        threshold = _resources.checklist()["thresholds"]["redundancy_similarity_max"]
    cons = _list(T, "constructs")
    vecs = [(c.get("id", ""), embedder(c.get("definition", ""))) for c in cons]
    rows = []
    for i in range(len(vecs)):
        for j in range(i + 1, len(vecs)):
            sim = rnd(_cosine(vecs[i][1], vecs[j][1]), 6)
            rows.append({
                "a": vecs[i][0], "b": vecs[j][0],
                "cosine": sim, "flag": "review" if sim >= threshold else "ok",
            })
    rows.sort(key=lambda r: (-r["cosine"], r["a"], r["b"]))
    return rows

render_report(T, path, title=None, render=False, to='html')

Write a Quarto report for the theory; render it with Quarto when render=True.

Returns the path of the written .qmd.

Source code in src/theoryforge/report_render.py

def render_report(T, path, title: str | None = None, render: bool = False, to: str = "html") -> str:
    """Write a Quarto report for the theory; render it with Quarto when ``render=True``.

    Returns the path of the written `.qmd`.
    """
    data = T.data if hasattr(T, "data") else T
    # Fall back to the id when the title is empty as well as absent (matches R's nzchar fallback).
    title = title or f"theoryforge report: {data.get('title') or data.get('id') or ''}"
    title = title.replace('"', "'")
    path = Path(path)
    if path.suffix.lower() != ".qmd":
        path = path.with_suffix(".qmd")
    header = f'---\ntitle: "{title}"\nformat: {to}\n---\n\n'
    path.write_text(header + _dossier(data), encoding="utf-8")
    if render:
        subprocess.run(["quarto", "render", str(path), "--to", to], check=True)  # pragma: no cover
    return str(path)

render_diagram(x, type='nomological_net')

Render a digraph view without leaving Python.

Where :func:theoryforge.diagram.diagram returns the deterministic Graphviz DOT string, render_diagram wraps it in a graphviz.Source, which displays inline in Jupyter and renders to a file with its render method. Requires the optional graphviz <https://pypi.org/project/graphviz/>_ library (pip install theoryforge[render]) and, to write image files, the Graphviz system binaries.

The three chart views (venn, rigour and severity) are already SVG, so they come back as an :class:SVGString, which also displays inline in Jupyter. The causal_dag view emits dagitty syntax rather than DOT, so it is not rendered here; paste it into a dagitty tool instead.

Parameters

x: A :class:~theoryforge.core.Theory, a parsed theory mapping, or a diagram IR string from diagram() or lit_diagram(), so literature diagrams render the same way. type: The diagram type, as in diagram(). Ignored when x is already an IR string.

Returns

graphviz.Source or SVGString A graphviz.Source for the digraph views; an :class:SVGString for the chart views.

Source code in src/theoryforge/render.py

def render_diagram(x, type: str = "nomological_net"):
    """Render a digraph view without leaving Python.

    Where :func:`theoryforge.diagram.diagram` returns the deterministic Graphviz
    DOT string, ``render_diagram`` wraps it in a ``graphviz.Source``, which
    displays inline in Jupyter and renders to a file with its ``render`` method.
    Requires the optional `graphviz <https://pypi.org/project/graphviz/>`_
    library (``pip install theoryforge[render]``) and, to write image files, the
    Graphviz system binaries.

    The three chart views (``venn``, ``rigour`` and ``severity``) are already
    SVG, so they come back as an :class:`SVGString`, which also displays inline
    in Jupyter. The ``causal_dag`` view emits dagitty syntax rather than DOT, so
    it is not rendered here; paste it into a dagitty tool instead.

    Parameters
    ----------
    x:
        A :class:`~theoryforge.core.Theory`, a parsed theory mapping, or a
        diagram IR string from ``diagram()`` or ``lit_diagram()``, so literature
        diagrams render the same way.
    type:
        The diagram type, as in ``diagram()``. Ignored when ``x`` is already an
        IR string.

    Returns
    -------
    graphviz.Source or SVGString
        A ``graphviz.Source`` for the digraph views; an :class:`SVGString` for
        the chart views.
    """
    if isinstance(x, str):
        ir = x
        if ir.lstrip().startswith("<svg"):
            return SVGString(ir)
        if ir.lstrip().startswith("dag"):
            raise ValueError(
                "this is dagitty syntax (the causal_dag view), not Graphviz DOT; "
                "render it with a dagitty tool such as dagitty.net."
            )
    else:
        if type == "causal_dag":
            raise ValueError(
                "the causal_dag view emits dagitty syntax, not Graphviz DOT; "
                "render diagram(x, 'causal_dag') with a dagitty tool such as "
                "dagitty.net."
            )
        if type in _SVG_TYPES:
            return SVGString(_diagram(x, type=type))
        if type not in _DOT_TYPES:
            # Delegate to diagram() for its canonical unknown-type error.
            _diagram(x, type=type)
        ir = _diagram(x, type=type)

    try:
        import graphviz
    except ImportError as err:
        raise ImportError(
            "render_diagram() needs the optional 'graphviz' library; install it "
            "with pip install theoryforge[render], or use diagram() and render "
            "the DOT string with any Graphviz tool."
        ) from err
    return graphviz.Source(ir)

osf_push(T, token=None, node=None, filename=None, dry_run=True, base_url=_DEFAULT_BASE)

Deposit the theory's dossier on OSF.

With dry_run=True (default) the planned request is returned and nothing is sent. A live upload (dry_run=False) requires both token and node (the OSF project id).

Source code in src/theoryforge/osf.py

def osf_push(T, token: str | None = None, node: str | None = None,
             filename: str | None = None, dry_run: bool = True,
             base_url: str = _DEFAULT_BASE) -> dict:
    """Deposit the theory's dossier on OSF.

    With ``dry_run=True`` (default) the planned request is returned and nothing is sent. A live
    upload (``dry_run=False``) requires both ``token`` and ``node`` (the OSF project id).
    """
    data = T.data if hasattr(T, "data") else T
    tid = data.get("id", "theory")
    fname = filename or f"{tid}.dossier.md"
    content = _dossier(data)
    # Percent-encode the filename (theory ids are user-supplied, so fname may
    # carry spaces, '&' or '#'); mirrors the R utils::URLencode(reserved = TRUE)
    # call so the dry-run request dicts stay parity-identical.
    url = (
        f"{base_url}{node}/providers/osfstorage/?kind=file&name={_quote(fname, safe='')}"
        if node else None
    )
    request = {"method": "PUT", "url": url, "filename": fname,
               "content_bytes": len(content.encode("utf-8"))}

    if dry_run:
        return {"dry_run": True, "request": request,
                "note": "set dry_run=False with a valid token and node to perform the upload"}
    if not token or not node:
        raise ValueError("a live OSF push requires both `token` and `node` (the OSF project id)")
    assert url is not None  # node is set, so url was constructed above
    import urllib.request  # pragma: no cover - network path, not exercised in tests
    req = urllib.request.Request(
        url, data=content.encode("utf-8"), method="PUT",
        headers={"Authorization": f"Bearer {token}", "Content-Type": "text/markdown"},
    )
    with urllib.request.urlopen(req, timeout=30) as resp:  # pragma: no cover - network
        return {"dry_run": False, "status": getattr(resp, "status", None), "filename": fname}

theoryforge.diagram.diagram(T, type='nomological_net', engine='graphviz')

Return the diagram IR string for the requested type.

engine is accepted but has no effect, because the IR is engine-independent (DOT for the digraphs, dagitty syntax for the causal DAG, and SVG for the Venn, the rigour grid and the severity chart).

Source code in src/theoryforge/diagram.py

def diagram(T: dict, type: str = "nomological_net", engine: str = "graphviz") -> str:
    """Return the diagram IR string for the requested type.

    ``engine`` is accepted but has no effect, because the IR is engine-independent
    (DOT for the digraphs, dagitty syntax for the causal DAG, and SVG for the Venn,
    the rigour grid and the severity chart).
    """
    T = T.data if hasattr(T, "data") else T
    if type == "nomological_net":
        return _nomological_net(T)
    if type == "provenance":
        return _provenance(T)
    if type == "causal_dag":
        return _causal_dag(T)
    if type == "development_roadmap":
        return _development_roadmap(T)
    if type == "pipeline":
        return _pipeline(T)
    if type == "context":
        return _context(T)
    if type == "workflow":
        return _workflow(T)
    if type == "venn":
        return _venn(T)
    if type == "rigour":
        return _rigor(T)
    if type == "severity":
        return _severity_chart(T)
    raise ValueError(f"unknown diagram type {type!r}; expected one of {_TYPES}")

theoryforge.dossier.dossier(T)

Source code in src/theoryforge/dossier.py

def dossier(T) -> str:
    data = T.data if hasattr(T, "data") else T
    rep = _check(data)
    lines = [
        f"# theoryforge dossier: {data.get('title', '')}",
        "",
        f"- Theory ID: {data.get('id', '')}",
        f"- Maturity: {data.get('maturity', '')}",
        f"- Aggregate rigour score: {_fmt(rep['aggregate_score'])}/100",
        f"- Gate: {rep['gate']}",
        f"- Blockers failed: {rep['n_blockers_failed']}",
        "",
        "## Rigour checklist",
        "",
        "| item | status | score | weight |",
        "| --- | --- | --- | --- |",
    ]
    for it in rep["items"]:
        lines.append(f"| {it['id']} | {it['status']} | {_fmt(it['score'])} | {_fmt(it['weight'])} |")

    lines += ["", "## Severity", ""]
    sev = _severity(data)
    if not sev:
        lines.append("_No predictions specified._")
    else:
        for s in sev:
            pid, cs, rk = s["prediction_id"], _fmt(s["computed_severity"]), _fmt(s["risk_score"])
            lines.append(f"- {pid}: severity {cs}, risk {rk}")

    lines += ["", "## Provenance", ""]
    prov = _list(data, "provenance")
    if not prov:
        lines.append("_No provenance recorded._")
    else:
        for i, s in enumerate(prov, start=1):
            action = str(s.get("action", "") or "")
            detail = str(s.get("detail", "") or "")
            lines.append(f"{i}. {action}: {detail}" if detail.strip() else f"{i}. {action}")

    lines += ["", "## Preregistration", ""]
    return "\n".join(lines) + "\n" + _preregister(data)

theoryforge.simulate.simulate(T, steps=10, dt=0.1, k=1.0, damping=0.5, init=1.0)

Integrate the theory's construct network as a linear dynamical system.

Returns {states, dt, steps, trajectory}, where trajectory[t] is the state vector at step t (t = 0..steps), each value rounded to 6 decimals.

Source code in src/theoryforge/simulate.py

def simulate(T, steps: int = 10, dt: float = 0.1, k: float = 1.0,
             damping: float = 0.5, init: float = 1.0) -> dict:
    """Integrate the theory's construct network as a linear dynamical system.

    Returns {states, dt, steps, trajectory}, where trajectory[t] is the state vector at
    step t (t = 0..steps), each value rounded to 6 decimals.
    """
    T = T.data if hasattr(T, "data") else T
    states = [c.get("id") for c in _list(T, "constructs")]
    n = len(states)
    idx = {s: i for i, s in enumerate(states)}

    A = [[0.0] * n for _ in range(n)]
    for p in _list(T, "propositions"):
        f, t, rel = p.get("from"), p.get("to"), p.get("relation")
        if f in idx and t in idx:
            sign = 1.0 if rel in _POS else (-1.0 if rel in _NEG else 0.0)
            A[idx[t]][idx[f]] += sign * k

    X = [float(init)] * n
    traj = [[rnd(x, 6) for x in X]]
    for _ in range(steps):
        dX = [sum(A[i][j] * X[j] for j in range(n)) - damping * X[i] for i in range(n)]
        X = [X[i] + dt * dX[i] for i in range(n)]
        traj.append([rnd(x, 6) for x in X])

    return {"states": states, "dt": dt, "steps": steps, "trajectory": traj}