feat(trace,workflow): terminal trace timeline, DAG diagram, run waterfall

Author: ctrl-gaurav

effgen/cli/_main.py

@@ -1101,9 +1101,25 @@ def run_agent(self, args):
                     if not quiet:
                         _progress.print_summary(self, response)
 
-                    # Display explain trace (tool reasoning)
-                    if getattr(args, 'explain', False) and response.execution_trace:
-                        self.print_header("Execution Trace (Explain Mode)")
+                    _explain = getattr(args, 'explain', False)
+                    _trace = getattr(args, 'trace', False)
+
+                    # A per-step timeline (bars + durations) shows where the
+                    # wall-clock went across the run's steps.
+                    if _trace and response.execution_trace:
+                        self.print_header("Timeline")
+                        _tl = _progress.execution_timeline_lines(response.execution_trace)
+                        if not _tl:
+                            self.print("(no timed steps recorded for this run)")
+                        for _style, _text in _tl:
+                            if self.console:
+                                self.console.print(f"[{_style}]{_text}[/{_style}]")
+                            else:
+                                print(_text)
+
+                    # Display the step trace (tool reasoning + per-step timing).
+                    if (_explain or _trace) and response.execution_trace:
+                        self.print_header("Execution Trace")
                         _lines = _progress.execution_trace_lines(response.execution_trace)
                         if not _lines:
                             self.print("(no detailed steps recorded for this run)")
@@ -1113,8 +1129,18 @@ def run_agent(self, args):
                             else:
                                 print(_text)
 
+                    # On a multi-step run without an explicit trace flag, point
+                    # the user at the timeline rather than leaving it hidden.
+                    elif not quiet and not _explain and int(getattr(response, "tool_calls", 0) or 0) >= 1:
+                        _steps = int(getattr(response, "tool_calls", 0) or 0)
+                        _hint = f"{_steps} tool step{'s' if _steps != 1 else ''} — run with --trace to see the timeline"
+                        if self.console:
+                            self.console.print(f"[effgen.muted]{_hint}[/effgen.muted]")
+                        else:
+                            print(_hint)
+
                     # Display execution statistics
-                    if getattr(args, 'verbose', False) or getattr(args, 'explain', False):
+                    if getattr(args, 'verbose', False) or _explain or _trace:
                         self.print_header("Execution Statistics")
                         stats_table = self._create_stats_table({
                             "Mode": response.mode.value,
@@ -2947,6 +2973,8 @@ def create_parser():
     )
     run_parser.add_argument('--explain', action='store_true',
                             help='Show why the agent chose each tool')
+    run_parser.add_argument('--trace', action='store_true',
+                            help='Show a step-by-step timeline with per-step durations')
     run_parser.add_argument('--checkpoint-dir', help='Directory to write agent checkpoints')
     run_parser.add_argument('--checkpoint-interval', type=int, default=0,
                             help='Checkpoint every N iterations (requires --checkpoint-dir)')
@@ -3254,11 +3282,20 @@ def create_parser():
                               'workflow entry node(s) (alternative to --input)')
     workflow_run.add_argument('--json', dest='output_json', action='store_true',
                               help='Emit the workflow result as JSON to stdout (for CI gating)')
+    workflow_run.add_argument('--diagram', action='store_true',
+                              help='Draw the workflow as a dependency graph (nodes by '
+                                   'level, edges, per-node status/duration/cost)')
+    workflow_run.add_argument('-q', '--quiet', action='store_true', default=argparse.SUPPRESS,
+                              help='Quiet output (errors only); --json still emits to stdout')
 
     workflow_validate = workflow_subparsers.add_parser('validate', help='Validate a workflow YAML file')
     workflow_validate.add_argument('file', help='Path to workflow YAML file')
     workflow_validate.add_argument('--json', dest='output_json', action='store_true',
                                    help='Emit the validation result as JSON to stdout')
+    workflow_validate.add_argument('--diagram', action='store_true',
+                                   help='Draw the workflow dependency graph (nodes by level, edges)')
+    workflow_validate.add_argument('-q', '--quiet', action='store_true', default=argparse.SUPPRESS,
+                                   help='Quiet output (errors only); --json still emits to stdout')
 
     # Batch command
     batch_parser = subparsers.add_parser(
@@ -4129,6 +4166,24 @@ def _handle_workflow_command(args, cli) -> int:
     json_mode = getattr(args, 'output_json', False)
     if json_mode:
         cli._human_to_stderr = True
+    show_diagram = getattr(args, 'diagram', False)
+
+    def _print_diagram(dag, node_results=None):
+        from effgen.ui.workflow_viz import workflow_diagram_lines
+        order = dag.topological_order()
+        levels = dag._compute_levels(order)
+        lines = workflow_diagram_lines(
+            dag.name,
+            [n.id for n in dag.nodes],
+            [e.to_dict() for e in dag.edges],
+            levels,
+            node_results=node_results,
+        )
+        for style, text in lines:
+            if cli.console and style:
+                cli.console.print(f"[{style}]{text}[/{style}]")
+            else:
+                cli.print(text)
 
     if wf_cmd == 'validate':
         try:
@@ -4147,6 +4202,9 @@ def _handle_workflow_command(args, cli) -> int:
             cli.print(f"  Nodes: {len(dag.nodes)}")
             cli.print(f"  Edges: {len(dag.edges)}")
             cli.print(f"  Execution order: {' -> '.join(order)}")
+            if show_diagram:
+                cli.print("")
+                _print_diagram(dag)
             return 0
         except Exception as e:
             if json_mode:
@@ -4197,8 +4255,10 @@ def _agent_factory(nd):
                 )
                 return Agent(config)
 
+            quiet = getattr(args, 'quiet', False)
             dag = WorkflowDAG.from_yaml(args.file, agent_factory=_agent_factory)
-            cli.print(f"Running workflow '{dag.name}' ({len(dag.nodes)} nodes)...")
+            if not quiet:
+                cli.print(f"Running workflow '{dag.name}' ({len(dag.nodes)} nodes)...")
 
             # Per-node ``task:`` strings declared in the YAML become each node's
             # default input (so `effgen workflow run workflow.yaml` works with no
@@ -4238,17 +4298,23 @@ def _agent_factory(nd):
                 print(json.dumps(result.to_dict(), indent=2, default=str, ensure_ascii=False))
                 return 0 if result.success else 1
 
-            cli.print(f"\nWorkflow {'succeeded' if result.success else 'FAILED'} "
-                       f"in {result.execution_time:.2f}s")
-            for nr in result.node_results:
-                status = nr['status']
-                cli.print(f"  [{status:>9s}] {nr['id']} ({nr['execution_time']:.2f}s)")
-
-            if result.success:
-                # Show final outputs
-                cli.print("\nOutputs:")
-                for key, val in result.outputs.items():
-                    cli.print(f"  {key}: {str(val)[:200]}")
+            if not quiet:
+                cli.print(f"\nWorkflow {'succeeded' if result.success else 'FAILED'} "
+                          f"in {result.execution_time:.2f}s")
+
+                if show_diagram:
+                    cli.print("")
+                    _print_diagram(dag, node_results=result.node_results)
+                else:
+                    for nr in result.node_results:
+                        status = nr['status']
+                        cli.print(f"  [{status:>9s}] {nr['id']} ({nr['execution_time']:.2f}s)")
+
+                if result.success:
+                    # Show final outputs
+                    cli.print("\nOutputs:")
+                    for key, val in result.outputs.items():
+                        cli.print(f"  {key}: {str(val)[:200]}")
 
             return 0 if result.success else 1
 

effgen/cli/progress.py

@@ -417,18 +417,50 @@ def format_tool_call(name: str, tool_input: Any, limit: int = 72) -> str:
     return f"{name}({inner})"
 
 
+def _fmt_duration(seconds: float) -> str:
+    """Human duration: ``820ms`` under a second, ``14.4s`` / ``1m03s`` above."""
+    if seconds < 1.0:
+        return f"{seconds * 1000:.0f}ms"
+    if seconds < 60.0:
+        return f"{seconds:.1f}s"
+    m, s = divmod(int(round(seconds)), 60)
+    return f"{m}m{s:02d}s"
+
+
+def _event_gaps(trace: list[dict[str, Any]]) -> list[float]:
+    """Wall-clock seconds elapsed *before* each event (gap from the previous one).
+
+    The gap before a ``tool_call_start`` is the model's think time for that
+    step — the time the flat trace throws away. The first event's gap is 0.
+    """
+    gaps: list[float] = []
+    prev: float | None = None
+    for ev in trace:
+        ts = ev.get("timestamp")
+        if isinstance(ts, int | float) and prev is not None:
+            gaps.append(max(0.0, float(ts) - prev))
+        else:
+            gaps.append(0.0)
+        if isinstance(ts, int | float):
+            prev = float(ts)
+    return gaps
+
+
 def execution_trace_lines(trace: list[dict[str, Any]] | None) -> list[tuple[str, str]]:
     """Turn an :class:`ExecutionTracker` event trace into readable ``(style, text)``.
 
     The agent's ``execution_trace`` is a list of event dicts (``type``,
     ``message``, ``data``) — not a ReAct ``thought/action/observation`` shape —
     so a plain ``step.get("thought")`` renders blank. This formatter walks the
     events and produces one human line each (reasoning, tool call + result,
-    delegation, …), shared by ``effgen run --explain`` and chat's ``/trace`` so
-    they agree.
+    delegation, …), annotating each step with the wall-clock time it took, and
+    is shared by ``effgen run --explain`` and chat's ``/trace`` so they agree.
     """
+    events = list(trace or [])
+    gaps = _event_gaps(events)
+    # Pair each tool_call_start with its terminal event to time the tool itself.
     out: list[tuple[str, str]] = []
-    for ev in trace or []:
+    for i, ev in enumerate(events):
         etype = str(ev.get("type", "") or "")
         msg = str(ev.get("message", "") or "")
         data = ev.get("data") or {}
@@ -441,10 +473,16 @@ def execution_trace_lines(trace: list[dict[str, Any]] | None) -> list[tuple[str,
                 detail = f"🔧 {format_tool_call(name, tool_input)}"
             else:
                 detail = f"🔧 {name}"
+            # Attribute the model's think time (the gap into this call) so the
+            # step reads with the time it cost, not as if it were instant.
+            if gaps[i] >= 0.1:
+                detail += f"  ⏱ {_fmt_duration(gaps[i])}"
             out.append(("green", detail))
         elif etype == "tool_call_complete":
             result = data.get("result", data.get("output", ""))
-            out.append(("dim", f"   ✓ {_truncate(result, 120)}" if result else "   ✓ done"))
+            exec_s = gaps[i] if gaps[i] >= 0.5 else 0.0
+            suffix = f"  ({_fmt_duration(exec_s)})" if exec_s else ""
+            out.append(("dim", f"   ✓ {_truncate(result, 120)}{suffix}" if result else f"   ✓ done{suffix}"))
         elif etype == "tool_call_failed":
             err = data.get("error", msg)
             out.append(("red", f"   ✗ {_truncate(err, 120)}"))
@@ -459,6 +497,52 @@ def execution_trace_lines(trace: list[dict[str, Any]] | None) -> list[tuple[str,
     return out
 
 
+def execution_timeline_lines(trace: list[dict[str, Any]] | None) -> list[tuple[str, str]]:
+    """A compact per-step timeline: each step with a proportional bar + duration.
+
+    Collapses the event stream into one line per meaningful step (a tool call,
+    reasoning, or delegation), sizes a bar by the step's wall-clock share, and
+    labels it with the elapsed time — so a run's slow steps are visible at a
+    glance. Falls back to an empty list when there is nothing timed to show.
+    """
+    events = list(trace or [])
+    if not events:
+        return []
+    gaps = _event_gaps(events)
+
+    # Build steps: a tool call is timed by the think-gap leading into it; a bare
+    # reasoning or delegation step is timed by its own leading gap.
+    steps: list[tuple[str, str, float]] = []  # (style, label, seconds)
+    for i, ev in enumerate(events):
+        etype = str(ev.get("type", "") or "")
+        data = ev.get("data") or {}
+        if etype == "tool_call_start":
+            name = data.get("tool_name", "tool")
+            tool_input = data.get("tool_input", data.get("input", ""))
+            label = format_tool_call(name, tool_input) if tool_input else name
+            steps.append(("green", f"🔧 {label}", gaps[i]))
+        elif etype == "sub_agent_start":
+            name = data.get("agent_name") or ev.get("agent_id") or "sub-agent"
+            steps.append(("magenta", f"👥 {name}", gaps[i]))
+        elif etype == "task_decomposition":
+            steps.append(("yellow", "🧩 planning", gaps[i]))
+    if not steps:
+        return []
+
+    longest = max((s for _, _, s in steps), default=0.0) or 1.0
+    bar_w = 20
+    label_w = min(40, max(len(lbl) for _, lbl, _ in steps))
+    out: list[tuple[str, str]] = []
+    for style, label, secs in steps:
+        filled = int(round((secs / longest) * bar_w)) if longest > 0 else 0
+        bar = "█" * filled + "·" * (bar_w - filled)
+        lbl = label if len(label) <= label_w else label[: label_w - 1] + "…"
+        out.append((style, f"{lbl:<{label_w}}  {bar}  {_fmt_duration(secs):>7}"))
+    total = sum(s for _, _, s in steps)
+    out.append(("dim", f"{'total':<{label_w}}  {'':<{bar_w}}  {_fmt_duration(total):>7}"))
+    return out
+
+
 # ---------------------------------------------------------------------------
 # Progress bars for the long-running, countable commands
 # ---------------------------------------------------------------------------

effgen/core/agent.py

@@ -1320,7 +1320,6 @@ def run(self,
                 # Add execution metadata
                 response.execution_time = time.time() - start_time
                 response.execution_trace = self.execution_tracker.get_trace()
-                response.execution_tree = self.execution_tracker.generate_execution_tree()
                 response.metadata["run_id"] = run_id
                 if input_redaction is not None:
                     response.metadata["input_redaction"] = input_redaction
@@ -1345,6 +1344,11 @@ def run(self,
                     }
                 ))
 
+                # Build the execution tree after completion is tracked so its
+                # root carries a terminal status and a real duration (not a
+                # still-"running" snapshot measured at serialize time).
+                response.execution_tree = self.execution_tracker.generate_execution_tree()
+
                 # Metrics: record latency and tokens
                 prom_metrics.response_latency.observe(response.execution_time, labels=labels)
                 if response.tokens_used:
@@ -1474,6 +1478,7 @@ def run(self,
                     success=False,
                     execution_time=time.time() - start_time,
                     execution_trace=self.execution_tracker.get_trace(),
+                    execution_tree=self.execution_tracker.generate_execution_tree(),
                     metadata={"reason": "run_failed", "error": detail, "run_id": run_id}
                 )
                 self._record_dashboard_run(response, error=redacted_msg)