project_state

project_state

Project state rendering widget and helpers.

ProjectState(**kwargs)

Bases: Static

Panel showing detailed information about the active project.

Initialize the project state panel.

Source code in src/terok/tui/widgets/project_state.py

def __init__(self, **kwargs: Any) -> None:
    """Initialize the project state panel."""
    super().__init__(**kwargs)

show_loading(project, task_count=None)

Show a loading placeholder while project state is being fetched.

Named show_loading (not set_loading) to avoid clashing with textual.Widget's identically-named bool toggle — different semantics.

Source code in src/terok/tui/widgets/project_state.py

def show_loading(self, project: ProjectConfig | None, task_count: int | None = None) -> None:
    """Show a loading placeholder while project state is being fetched.

    Named *show_loading* (not *set_loading*) to avoid clashing with
    textual.Widget's identically-named bool toggle — different semantics.
    """
    self.update(render_project_loading(project, task_count))

set_state(project, state, task_count=None, staleness=None, shield_env=None, review_lag=None)

Display fully loaded project details including infrastructure status.

Source code in src/terok/tui/widgets/project_state.py

def set_state(
    self,
    project: ProjectConfig | None,
    state: dict | None,
    task_count: int | None = None,
    staleness: GateStalenessInfo | None = None,
    shield_env: EnvironmentCheck | None = None,
    review_lag: list[str] | None = None,
) -> None:
    """Display fully loaded project details including infrastructure status."""
    self.update(
        render_project_details(
            project,
            state,
            task_count,
            staleness,
            _get_css_variables(self),
            shield_env=shield_env,
            review_lag=review_lag,
        )
    )

set_broken(bp)

Show bp's validation error in place of the normal infrastructure readout.

Source code in src/terok/tui/widgets/project_state.py

def set_broken(self, bp: BrokenProject) -> None:
    """Show *bp*'s validation error in place of the normal infrastructure readout."""
    self.update(render_broken_project(bp, _get_css_variables(self)))

render_project_loading(project, task_count=None)

Render project loading state as a Rich Text object.

Source code in src/terok/tui/widgets/project_state.py

def render_project_loading(
    project: ProjectConfig | None,
    task_count: int | None = None,
) -> Text:
    """Render project loading state as a Rich Text object."""
    if project is None:
        return Text("No project selected.")

    upstream = project.upstream_url or "-"
    sec = SECURITY_CLASS_DISPLAY.get(project.security_class, SECURITY_CLASS_DISPLAY["online"])
    gpu = GPU_DISPLAY[has_gpu(project)]
    badges = f"{render_emoji(sec)}{render_emoji(gpu)}"
    tasks_line = (
        Text("Tasks:     loading") if task_count is None else Text(f"Tasks:     {task_count}")
    )

    lines = [Text(f"Project:   {project.name} {badges}")]
    if project.description:
        lines.append(Text(f"Desc:      {project.description}", style=Style(dim=True)))
    lines += [
        Text(upstream),
        Text(""),
        Text("Loading details..."),
        tasks_line,
    ]
    return Text("\n").join(lines)

render_broken_project(bp, css_variables=None)

Render a damaged project's validation error as a Rich Text block (#565).

The config path plus the raw error message are enough for the user to open project.yml in an editor and fix whatever pydantic or the YAML parser flagged — no extra round-trip to the CLI.

Source code in src/terok/tui/widgets/project_state.py

def render_broken_project(bp: BrokenProject, css_variables: dict[str, str] | None = None) -> Text:
    """Render a damaged project's validation error as a Rich Text block (#565).

    The config path plus the raw error message are enough for the user to
    open ``project.yml`` in an editor and fix whatever pydantic or the
    YAML parser flagged — no extra round-trip to the CLI.
    """
    variables = css_variables or {}
    error_color = variables.get("error", "red")
    dim = Style(dim=True)
    header = Text(f"Project:   {bp.name} ", style=Style(color=error_color, bold=True))
    header.append("(broken)", style=Style(color=error_color))
    lines = [
        header,
        Text(""),
        Text("This project's configuration could not be loaded:"),
        Text(""),
        Text(bp.error, style=Style(color=error_color)),
        Text(""),
        Text(f"Config: {bp.config_path}", style=dim),
    ]
    return Text("\n").join(lines)

render_project_details(project, state, task_count=None, staleness=None, css_variables=None, shield_env=None, review_lag=None)

Render project details as a Rich Text object.

Source code in src/terok/tui/widgets/project_state.py

def render_project_details(
    project: ProjectConfig | None,
    state: dict | None,
    task_count: int | None = None,
    staleness: GateStalenessInfo | None = None,
    css_variables: dict[str, str] | None = None,
    shield_env: EnvironmentCheck | None = None,
    review_lag: list[str] | None = None,
) -> Text:
    """Render project details as a Rich Text object."""
    if project is None or state is None:
        return Text("No project selected.")

    variables = css_variables or {}
    success_color = variables.get("success", "green")
    error_color = variables.get("error", "red")
    warning_color = variables.get("warning", "yellow")

    status_styles = {
        "yes": Style(color=success_color),
        "no": Style(color=error_color),
        "old": Style(color=warning_color),
        "new": Style(color="blue"),
    }

    def _status_text(value: str) -> Text:
        """Return a styled Rich Text for a status value like 'yes', 'no', or 'old'."""
        style = status_styles.get(value, Style(color=error_color))
        return Text(value, style=style)

    docker_value = "yes" if state.get("dockerfiles") else "no"
    if docker_value == "yes" and state.get("dockerfiles_old"):
        docker_value = "old"
    docker_s = _status_text(docker_value)

    images_value = "yes" if state.get("images") else "no"
    if images_value == "yes" and state.get("images_old"):
        stale = state.get("stale_layers", [])
        hint = _stale_layer_hint(stale)
        images_value = f"old {hint}" if hint else "old"
    images_s = _status_text(images_value)
    ssh_s = _status_text("yes" if state.get("ssh") else "no")

    # Gate line: per-project mirror existence + staleness vs upstream.
    # The gate runs inside each task container's supervisor now — there is
    # no host gate daemon whose status could override the mirror state.
    gate_value = "yes" if state.get("gate") else "no"
    if gate_value == "yes" and staleness is not None and not staleness.error and staleness.is_stale:
        behind = staleness.commits_behind or 0
        ahead = staleness.commits_ahead or 0
        if ahead > 0 and behind == 0:
            gate_value = "new"
        else:
            gate_value = "old"
    gate_s = _status_text(gate_value)
    # Pending destructive ops badge — persists until the operator resolves
    # them (press g), so a declined-or-unseen proposal can't just vanish.
    if pending_ops := state.get("gate_pending_ops"):
        gate_s = Text.assemble(
            gate_s, "  ", (f"{pending_ops} pending!", Style(color=error_color, bold=True))
        )

    tasks_line = (
        Text("Tasks:     unknown") if task_count is None else Text(f"Tasks:     {task_count}")
    )
    upstream = project.upstream_url or "-"
    sec = SECURITY_CLASS_DISPLAY.get(project.security_class, SECURITY_CLASS_DISPLAY["online"])
    gpu = GPU_DISPLAY[has_gpu(project)]
    badges = f"{render_emoji(sec)}{render_emoji(gpu)}"

    dim_style = Style(dim=True)
    # Three-state badge based on YAML config + file existence
    yaml_instructions = project.agent_config.get("instructions")
    try:
        has_file = (project.root / "instructions.md").is_file()
    except (TypeError, AttributeError) as exc:
        from terok.lib.util.logging_utils import _log_debug

        _log_debug(f"project_state: instructions.md existence probe failed: {exc}")
        has_file = False
    has_yaml = yaml_instructions is not None

    if has_yaml or has_file:
        # Check if _inherit is present (or YAML is absent = implicit inherit)
        inherits = not has_yaml
        if isinstance(yaml_instructions, list):
            inherits = "_inherit" in yaml_instructions
        elif isinstance(yaml_instructions, dict):
            inherits = any(
                isinstance(v, list) and "_inherit" in v for v in yaml_instructions.values()
            )
        if inherits:
            instr_s = Text("custom + inherited", style=Style(color=success_color))
        else:
            instr_s = Text("custom only", style=Style(color="cyan"))
    else:
        instr_s = Text("default", style=dim_style)

    # Shield status line
    if shield_env is not None:
        _shield_colors = {
            "ok": success_color,
            "setup-needed": error_color,
            "stale-hooks": warning_color,
            "disabled": error_color,
        }
        shield_color = _shield_colors.get(shield_env.health, error_color)
        shield_s = Text(shield_env.health, style=Style(color=shield_color))
    else:
        shield_s = Text("unknown", style=Style(dim=True))

    lines = [Text(f"Project:   {project.name} {badges}")]
    if project.description:
        lines.append(Text(f"Desc:      {project.description}", style=Style(dim=True)))
    lines += [
        Text(upstream),
        Text(""),
        Text.assemble("Dockerfiles: ", docker_s),
        Text.assemble("Images:      ", images_s),
        Text.assemble("SSH dir:     ", ssh_s),
        Text.assemble("Git gate:    ", gate_s),
        Text.assemble("Shield:      ", shield_s),
        Text.assemble("Instruct:    ", instr_s),
        tasks_line,
    ]

    gate_commit = state.get("gate_last_commit")
    if gate_commit:
        commit_hash = gate_commit.get("commit_hash") or "unknown"
        commit_hash_short = commit_hash[:8] if isinstance(commit_hash, str) else "unknown"
        commit_date = gate_commit.get("commit_date") or "unknown"
        commit_author = gate_commit.get("commit_author") or "unknown"
        commit_message = gate_commit.get("commit_message") or "unknown"
        commit_message_short = (
            commit_message[:50] + ("..." if len(commit_message) > 50 else "")
            if isinstance(commit_message, str)
            else "unknown"
        )

        lines.append(Text(""))
        lines.append(Text("Gate info:"))
        lines.append(Text(f"  Commit:   {commit_hash_short}"))
        lines.append(Text(f"  Date:     {commit_date}"))
        lines.append(Text(f"  Author:   {commit_author}"))
        lines.append(Text(f"  Message:  {commit_message_short}"))

    if staleness is not None:
        lines.append(Text(""))
        lines.append(Text("Upstream status:"))
        if staleness.error:
            lines.append(Text(f"  Error:    {staleness.error}"))
        elif staleness.is_stale:
            behind = staleness.commits_behind or 0
            ahead = staleness.commits_ahead or 0

            if ahead > 0 and behind > 0:
                status_str = f"DIVERGED ({ahead} ahead, {behind} behind) on {staleness.branch}"
            elif ahead > 0:
                status_str = f"AHEAD ({ahead} commits) on {staleness.branch}"
            else:
                behind_str = "unknown" if staleness.commits_behind is None else str(behind)
                status_str = f"BEHIND ({behind_str} commits) on {staleness.branch}"

            lines.append(Text(f"  Status:   {status_str}"))
            upstream_head = staleness.upstream_head[:8] if staleness.upstream_head else "unknown"
            gate_head = staleness.gate_head[:8] if staleness.gate_head else "unknown"
            lines.append(Text(f"  Upstream: {upstream_head}"))
            lines.append(Text(f"  Gate:     {gate_head}"))
        else:
            lines.append(Text(f"  Status:   Up to date on {staleness.branch}"))
            gate_head = staleness.gate_head[:8] if staleness.gate_head else "unknown"
            lines.append(Text(f"  Commit:   {gate_head}"))
        lines.append(Text(f"  Checked:  {staleness.last_checked}"))

    # Review lag — gate branches ahead of an open MR/PR (level-triggered:
    # shown until the operator pushes the gate or the review closes).
    if review_lag:
        lines.append(Text(""))
        lines.append(Text("Review lag (gate ahead of open review — push the gate!):"))
        lines.extend(
            Text(f"  {warning}", Style(color=warning_color, bold=True)) for warning in review_lag
        )

    lines.append(Text(""))
    lines.append(Text(f"Config: {project.root}", style=dim_style))

    return Text("\n").join(lines)