penguiflow

Measure cyclomatic complexity across a Python repo to find refactor opportunities and track code health over time. Runs three analyses — ruff C901 violations (functions above a CC threshold), radon repo-wide average and grade distribution (A–F), and a reference-weighted hotspot ranking that prioritizes complex code that's actually used. Use when the user asks about cyclomatic complexity, code complexity, where to refactor, repo hygiene metrics, complexity hotspots, technical-debt prioritization, complexity scores, or wants to add a complexity gate to CI/CD. Outputs human-readable tables plus an optional JSON report suitable for trend tracking and CI artifacts.

Configure PenguiFlow ReactPlanner in production codebases (native LLM adapter, tool discovery + deferred activation, skills packs, allowlists/visibility policies, rich output, A2A router integration, guardrails, and background tasks) with safe defaults and troubleshooting.

Author "skill packs" — playbooks the PenguiFlow ReactPlanner retrieves at runtime via `skill_search`/`skill_get`/`skill_list` (NOT the same as the SKILL.md agent-tool descriptors in the project's `skills/` directory; these are runtime artifacts stored in a SQLite skill store). Configure `SkillsConfig(enabled=True, skill_packs=[SkillPackConfig(...)], top_k=..., redact_pii=True)`, pick a pack format (`*.skill.md`/`.yaml`/`.json`/`.jsonl`), declare applicability via `required_tool_names`/`required_namespaces`/`required_tags`, scope by tenant/project, compose with a runtime `skills_provider`/`skills_provider_factory`, optionally enable non-persisting `skill_propose` drafting and `SkillsDirectoryConfig` surfacing. Use when a user says "author a skill pack", "playbooks for my planner", "skill_search", "runtime skills provider", or names `SkillsConfig`.

Explain, configure, and troubleshoot the ReactPlanner reflection loop: how post-answer critique, revision, and clarification work; how to wire `ReflectionConfig` and optional `reflection_llm`; how env/spec-driven scaffolds expose the knobs; and how to verify behavior through planner events, metadata, and cost counters. Use when a user asks about "reflection loop", "self-critique", "reflection_critique", "quality_threshold", "max_revisions", or wants to compare reflection-enabled vs reflection-disabled runs.

Deep expertise for PenguiFlow persistence: implement or review a `StateStore` backend (required + optional duck-typed capabilities), enforce correct contracts (idempotency, ordering, cursors, cancellation), wire A2A router continuity via conversation bindings, sessions/tasks/updates/steering and background-task orchestration, integrate `ArtifactStore`, support distributed execution (`RemoteBinding`), and build UIs/clients around the Playground HTTP+SSE contract. Use when building/debugging StateStore or ArtifactStore implementations; designing database schemas; investigating missing/duplicated/out-of-order events; integrating A2A manager/specialist continuity, OAuth/HITL pause-resume; persisting trajectories/planner events; or implementing frontends that consume `/chat`, `/session/stream`, `/tasks`, `/events`, `/trajectory`, and `/artifacts`.

Expose a PenguiFlow agent as an A2A-spec service or call remote A2A agents from a planner — wire `A2AService` to a FastAPI app via `install_a2a_http`/`create_a2a_http_router`/`create_a2a_http_app`, serve the agent card at `/.well-known/agent-card.json`, handle JSON-RPC + REST + SSE + push-notification transports, optionally enable the gRPC binding, build manager↔specialist topologies with `A2AAgentToolset` and an `AgentRegistry`, persist conversation continuity via `RemoteBinding`/`StateStore`, and choose blocking/stream/task execution modes. Use when a user says "expose as A2A", "agent-to-agent", "call a remote agent", "agent card", "manager/specialist", "JSON-RPC", "SSE for agents", "push notifications for agents", or names `A2AService`, `RemoteBinding`, `A2AAgentToolset`.

Clone an existing PenguiFlow test agent into a new folder, remove copied state, rewire MCP/tooling for a new idea, and validate it boots cleanly. Use when you want to spin up another experiment agent quickly from a working sibling — not for greenfield agents. For brand-new agents, prefer `penguiflow new <template>` (interactive) or `penguiflow generate <spec.yaml>` (declarative, recommended Jan 2026+); see `TEMPLATING_QUICKGUIDE.md`.

Enable concurrent background work on a PenguiFlow ReactPlanner — configure `BackgroundTasksConfig(enabled=True, allow_tool_background=..., default_mode=..., default_merge_strategy=..., max_concurrent_tasks=..., task_timeout_s=...)`, expose the `tasks.*` tool surface via `build_task_tool_specs()`, supply `tool_context["task_service"]` (implementing the `TaskService` protocol) plus `session_id`, choose merge strategies (`HUMAN_GATED`/`APPEND`/`REPLACE`), use task groups for coherent multi-task reports, and let tools opt-into background execution via `spec.extra["background"]`. Use when a user says "background task", "spawn a subagent", "async tool", "task groups", or names `BackgroundTasksConfig`, `tasks.spawn`, `task.subagent`.

Build deterministic async pipelines and agent graphs with the PenguiFlow core runtime — wire Nodes with NodePolicy, construct the graph with `create(...)` and adjacency tuples, use Messages with Headers/trace_id envelopes, run/emit/fetch/stop, branch with `predicate_router`/`union_router`, fan-out and `join_k`, run parallel work with `map_concurrent`, cancel a trace, set deadlines, retry transient failures, and surface FlowError to callers. Use when a user says "build a flow", "wire a node", "fan-out/fan-in", "router", "cancel a run", "set a deadline", "retry policy", "envelope vs payload", "queue backpressure", or asks how the runtime works without a planner.

Take a PenguiFlow agent from local dev to a production worker fleet — pick a deployment shape (embedded service, queue-backed worker pool, distributed), enforce envelope discipline (`Message` + `Headers.tenant` + `trace_id`), set runtime limits (`queue_maxsize`, `NodePolicy.timeout_s`/`max_retries`, `Message.deadline_s`), implement worker lifecycle (`flow.run(...)` → `await flow.stop()`), wire durability via `StateStore` capabilities, configure distributed hooks (`MessageBus`, `RemoteTransport`+`RemoteNode`, A2A bindings), and harden tool execution. Use when a user says "deploy to production", "worker integration", "multi-worker", "queue-backed", "distributed execution", "RemoteNode", "MessageBus", or asks how to size queues/timeouts/retries.

Add a policy-enforcement layer to a PenguiFlow ReactPlanner — pass a `guardrail_gateway` that inspects `llm_before`/`tool_call_start`/`tool_call_result`/`llm_stream_chunk` events and returns `GuardrailDecision`s with actions `STOP`/`PAUSE`/`RETRY`/`REDACT`/`ALLOW`, run in `shadow` or `enforce` mode, classify rules as FAST (sync, `sync_timeout_ms`) or DEEP (async, `SteeringGuardInbox`), choose `sync_fail_open` vs safety, and configure the separate `ObservationGuardrailConfig` reliability clamp that prevents context overflow on huge tool outputs. Use when a user says "add guardrails", "PII redaction", "tool denylist", "policy enforcement", "stop unsafe tool", "fail-closed", or names `GuardrailEvent`, `GuardrailDecision`, `GatewayConfig`.

Wire external tools (MCP servers, UTCP, HTTP, CLI) into a PenguiFlow ReactPlanner via `ToolNode` and `ExternalToolConfig` — pick a transport, choose an auth mode (none/API key/bearer/cookie/OAuth2), substitute env vars, set timeouts/retries/concurrency, allowlist tools with `tool_filter`, enable resource discovery (`{ns}.resources_list`/`resources_read`), extract large content into artifacts, and gate OAuth flows through HITL pause/resume. Use when a user says "wire MCP tools", "connect to an MCP server", "external tools", "ToolNode", "OAuth for tools", "tool filter", "UTCP", or names `ExternalToolConfig`, `TransportType`, `AuthType`.

Add short-term conversation memory to a PenguiFlow ReactPlanner — configure `ShortTermMemoryConfig` with strategy (`none`/`truncation`/`rolling_summary`), `MemoryBudget` token caps + overflow policy, `MemoryIsolation` for multi-tenant fail-closed scoping via `MemoryKey(tenant_id, user_id, session_id)`, optionally persist via `StateStore.save_memory_state`/`load_memory_state`, wire `on_turn_added`/`on_summary_updated`/`on_health_changed` hooks, and tune `summarizer_model`/`include_trajectory_digest`. Use when a user says "add memory", "session memory", "conversation memory", "multi-tenant memory", "rolling summary", or names `ShortTermMemoryConfig`, `MemoryKey`, `MemoryIsolation`.

Add structured observability to a PenguiFlow runtime — turn on JSON logging with `configure_logging(structured=True)`, attach `log_flow_events` middleware to capture `FlowEvent` lifecycle (`node_start`/`node_success`/`node_error`/`node_timeout`/`node_retry`/`node_failed`/`deadline_skip`/`trace_cancel_*`), derive counters/histograms/gauges via `FlowEvent.metric_samples()`/`tag_values()` while obeying cardinality rules (never tag by `trace_id`), wire `FlowError`/`FlowErrorCode` for traceable failures, and export topology with `flow_to_mermaid`/`flow_to_dot`. Use when a user says "add observability", "logging", "metrics", "alerts", "trace this flow", "visualize", or asks how to monitor a production worker fleet.

Add structured UI components (charts, tables, reports, grids, tabs, accordions, forms) to a PenguiFlow ReactPlanner — enable rich output via `attach_rich_output_nodes(registry, config=RichOutputConfig(enabled=True, allowlist=[...]))`, pick the right tool family (generic `render_component`, typed `render_*` wrappers like `render_chart_echarts`/`render_table`/`render_report`/`render_grid`/`render_tabs`/`render_accordion`, non-emitting `build_*` builders that return `artifact_ref`, interactive `ui_form`/`ui_confirm`/`ui_select_option`), describe schemas with `describe_component`, list artifacts with `list_artifacts`, and consume `artifact_chunk(artifact_type="ui_component")` on the frontend. Use when a user says "render a chart", "show a table", "rich output", "UI components", "interactive form", or names `render_component`, `RichOutputConfig`, `attach_rich_output_nodes`.

Write concise PenguiFlow unit tests with `penguiflow.testkit` — drive a single envelope trace through a flow with `run_one(flow, message, registry=None, timeout_s=1.0)`, assert execution order with `assert_node_sequence(trace_id, [...])`, simulate transient or terminal failures with `simulate_error(node_name, code, fail_times=N)` for retry/timeout tests, inspect captured events with `get_recorded_events(trace_id)`, and verify Message envelope preservation with `assert_preserves_message_envelope(node, ...)`. Use when a user says "write a flow test", "FlowTestKit", "test a node", "test retries", "simulate a failure", "assert node order", or asks how to unit-test PenguiFlow code with pytest-asyncio.

Implement, extend, and debug PenguiFlow's AG-UI (AGUI) event streaming contract across backend and frontend: FastAPI /agui/agent + /agui/resume endpoints, PenguiFlowAdapter mapping from ReactPlanner PlannerEvent to AG-UI events, custom events (artifact_stored, artifact_chunk, resource_updated, pause, state_update, thinking, revision), and frontend reducer/state machine patterns for smooth streaming UI with persistence/replay.

Add human-in-the-loop pause/resume to a PenguiFlow ReactPlanner — let tools call `ctx.pause(reason, payload)` to gate approvals, OAuth handoffs, await-input prompts, and policy decisions; consume the resulting `PlannerPause(reason, payload, resume_token)` in your UI; call `planner.resume(resume_token, user_input=..., tool_context=...)` to continue; and persist pause state across workers via a `StateStore` implementing `save_planner_state`/`load_planner_state`. Use when a user says "pause for HITL", "human in the loop", "approval gate", "wait for user input", "OAuth handoff", "resume after restart", or names `PlannerPause`, `ctx.pause`, `resume_token`.

Scaffold a new PenguiFlow agent project end-to-end using the CLI — choose the right template (minimal, react, parallel, flow, controller, rag_server, wayfinder, analyst, enterprise), pick the right `--with-*` flags (streaming, hitl, a2a, rich-output, background-tasks, --no-memory), run the playground, and iterate. Use when a user says "create a new agent", "scaffold", "start a PenguiFlow project", "which template should I use", "spec-driven agent", or asks about `penguiflow new`, `penguiflow generate`, `penguiflow dev`, `penguiflow init`. Covers both interactive scaffolding (`penguiflow new`) and declarative spec-first generation (`penguiflow generate`).

Stream partial output (LLM tokens, progressive synthesis, status updates) from a PenguiFlow flow — emit `StreamChunk` messages via `ctx.emit_chunk(parent=msg, text=..., done=..., to=sink)` from a "compose" node, route them to a dedicated sink node, and ship them to clients via SSE (`format_sse_event`), WebSocket (`chunk_to_ws_json`), or programmatic iteration (`stream_flow`/`emit_stream_events`). Use when a user says "stream tokens", "SSE streaming", "WebSocket streaming", "progressive output", "StreamChunk", or asks how to stream partials without the higher-level AG-UI protocol.

Design and critique user interfaces (web/mobile) using a calm, cozy-premium “Design Soul” aesthetic (warm parchment neutrals, muted mint/teal accents, soft rounded surfaces, subtle depth, and generous whitespace). Use for UX/UI direction, screen layouts, information architecture, component + state specs, design tokens (CSS/Tailwind), interaction/motion guidance, accessibility review, and implementation handoff notes.

Guide for authenticating with Databricks Apps using cookie-based auth when OAuth/PAT tokens don't work. Use when connecting to Databricks Apps with User Authorization enabled.

Guidelines for implementing generalizable solutions in the penguiflow library. Use when modifying library code, adding features, or fixing bugs in penguiflow core.

Workflow for propagating penguiflow library changes to test agents. Use when modifying library code and need to test changes in test_generation agents.