Skip to content

Command Line Interface (CLI) Reference 💻

This section documents the subcommands and command-line execution helper modules in dv-agentic-system.

Agent Invocation Commands

These CLI handlers map command-line arguments to agent runs.

Orchestrator Command

orchestrator

CLI entrypoint for OrchestratorAgent.

Sub-agents are wired up automatically based on --simulator and --adapter flags. The LLM client is selected from environment variables (see :mod:~dv_agentic.cli._factory).

When --project-config is provided, the three-layer configuration system is activated: team profile, IP-type rules, and adapter settings are all loaded from .agent/project.yaml and the profiles directory, and injected into every agent's system prompt via :class:~dv_agentic.prompts.prompt_loader.PromptLoader.

Examples:

.. code-block:: shell

# Minimal: no profile injection
python3 -m dv_agentic.cli.orchestrator \
    --input-file task.txt \
    --simulator xcelium \
    --adapter imc

# Full: load team + IP profiles from project.yaml
python3 -m dv_agentic.cli.orchestrator \
    --project-config .agent/project.yaml \
    --profiles-dir ../team-profiles \
    --input-file task.txt
main()

Main execution block of the Orchestrator CLI.

Source code in src/dv_agentic/cli/orchestrator.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def main() -> None:
    """Main execution block of the Orchestrator CLI."""
    args = _build_parser().parse_args()

    try:
        task_input = read_input(args.input_file)
    except OSError as exc:
        exit_with_error(str(exc))

    # --- Three-layer config ---
    project_ctx = None
    project_simulator = None
    project_coverage = None
    if args.project_config:
        try:
            from dv_agentic.config import load_project

            project_ctx, project_simulator, project_coverage = load_project(
                project_yaml=args.project_config,
                profiles_dir=args.profiles_dir,
            )
        except (FileNotFoundError, ValueError) as exc:
            exit_with_error(f"Failed to load project config: {exc}")

    try:
        from dv_agentic.agents.base import AgentConfig
        from dv_agentic.agents.orchestrator import OrchestratorAgent
        from dv_agentic.tools.adapters import get_coverage_adapter, get_simulator_adapter

        llm = make_llm(model=args.model)
        sub_agents = _build_sub_agents(args, llm, project_ctx=project_ctx)

        agent = OrchestratorAgent(
            config=AgentConfig(name="orchestrator", budget=args.budget),
            llm=llm,
            sub_agents=sub_agents,
            project_config=project_ctx,
            simulator=project_simulator or get_simulator_adapter(args.simulator),
            coverage=project_coverage or get_coverage_adapter(args.adapter),
            coverage_threshold=args.coverage_threshold,
            sim_max_runs=args.sub_budget,
        )
        result = asyncio.run(agent.run(task_input))
        print(result)  # noqa: T201
    except Exception as exc:
        exit_with_error(str(exc))

Code Generator Command

code_generator

CLI entrypoint for CodeGeneratorAgent.

Examples:

.. code-block:: shell

python3 -m dv_agentic.cli.code_generator             --task-id cov_fix_001             --input-file task_description.txt
main()

Main execution block of the CodeGenerator CLI.

Source code in src/dv_agentic/cli/code_generator.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def main() -> None:
    """Main execution block of the CodeGenerator CLI."""
    args = _build_parser().parse_args()

    if args.no_tb_guard:
        logger.warning(
            "============================================================\n"
            "SECURITY WARNING: Testbench guard is DISABLED (--no-tb-guard).\n"
            "The agent is allowed to write to any path in the workspace,\n"
            "including RTL source files. USE ONLY IN TEST ENVIRONMENTS!\n"
            "============================================================"
        )

    try:
        description = read_input(args.input_file)
    except OSError as exc:
        exit_with_error(str(exc))

    try:
        from dv_agentic.agents.base import AgentConfig
        from dv_agentic.agents.code_generator import (
            DEFAULT_TB_ALLOWED_DIRS,
            CodeGeneratorAgent,
            CodeTask,
        )

        llm = make_llm(model=args.model)
        agent = CodeGeneratorAgent(
            config=AgentConfig(name="code_generator", budget=args.budget),
            llm=llm,
            workspace_dir=".",
            # TB guard is ON by default; --no-tb-guard disables it
            allowed_dirs=None if args.no_tb_guard else DEFAULT_TB_ALLOWED_DIRS,
        )
        task = CodeTask(task_id=args.task_id, description=description)
        result = asyncio.run(agent.run(task))
        print(result)  # noqa: T201
    except Exception as exc:
        exit_with_error(str(exc))

Sim Controller Command

sim_controller

CLI entrypoint for SimControllerAgent.

Examples:

.. code-block:: shell

python3 -m dv_agentic.cli.sim_controller             --task-id cov_fix_001             --test axi_burst_test             --seed 42             --simulator xcelium
main()

Main execution block of the SimController CLI.

Source code in src/dv_agentic/cli/sim_controller.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def main() -> None:
    """Main execution block of the SimController CLI."""
    args = _build_parser().parse_args()

    try:
        from dv_agentic.agents.base import AgentConfig
        from dv_agentic.agents.sim_controller import SimControllerAgent
        from dv_agentic.tools.adapters import get_simulator_adapter
        from dv_agentic.tools.models import SimTask

        simulator = get_simulator_adapter(args.simulator)
        agent = SimControllerAgent(
            config=AgentConfig(name="sim_controller", budget=args.budget),
            simulator=simulator,
            base_branch=args.base_branch,
        )
        task = SimTask(
            task_id=args.task_id,
            test=args.test,
            seed=args.seed,
            file_list=args.file_list,
            top=args.top,
            debug=args.debug,
        )
        result = asyncio.run(agent.run(task))
        print(result)  # noqa: T201
    except Exception as exc:
        exit_with_error(str(exc))

Log Analyzer Command

log_analyzer

CLI entrypoint for LogAnalyzerAgent.

Examples:

.. code-block:: shell

python3 -m dv_agentic.cli.log_analyzer --input-file sim_test_42.log
python3 -m dv_agentic.cli.log_analyzer --input-file -         # read stdin
cat sim.log | python3 -m dv_agentic.cli.log_analyzer
main()

Main execution block of the LogAnalyzer CLI.

Source code in src/dv_agentic/cli/log_analyzer.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def main() -> None:
    """Main execution block of the LogAnalyzer CLI."""
    args = _build_parser().parse_args()

    try:
        content = read_input(args.input_file)
    except OSError as exc:
        exit_with_error(str(exc))

    from dv_agentic.agents.base import AgentConfig
    from dv_agentic.agents.log_analyzer import LogAnalyzerAgent

    agent = LogAnalyzerAgent(config=AgentConfig(name="log_analyzer"))
    result = asyncio.run(agent.run(content))
    print(result)  # noqa: T201

Specialty Commands

Spec Analyst Command

spec_analyst

CLI entrypoint for SpecAnalystAgent.

Examples:

.. code-block:: shell

python3 -m dv_agentic.cli.spec_analyst --input-file spec.txt --output-path .agent/vplan.yaml
main()

Main execution block of the SpecAnalyst CLI.

Source code in src/dv_agentic/cli/spec_analyst.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def main() -> None:
    """Main execution block of the SpecAnalyst CLI."""
    args = _build_parser().parse_args()

    try:
        spec_text = read_input(args.input_file)
    except OSError as exc:
        exit_with_error(str(exc))

    output_path = args.output_path

    try:
        from dv_agentic.agents.base import AgentConfig
        from dv_agentic.agents.spec_analyst import SpecAnalystAgent

        llm = make_llm(model=args.model)
        agent = SpecAnalystAgent(
            config=AgentConfig(name="spec_analyst", budget=args.budget),
            llm=llm,
            output_path=output_path,
        )
        result = asyncio.run(agent.run(spec_text))
        print(result)  # noqa: T201
    except Exception as exc:
        exit_with_error(str(exc))

Bug Classifier Command

bug_classifier

CLI entrypoint for BugClassifierAgent.

Examples:

.. code-block:: shell

python3 -m dv_agentic.cli.bug_classifier --input-file failure_summary.txt --threshold 0.75
cat failure.txt | python3 -m dv_agentic.cli.bug_classifier
main()

Main execution block of the BugClassifier CLI.

Source code in src/dv_agentic/cli/bug_classifier.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def main() -> None:
    """Main execution block of the BugClassifier CLI."""
    args = _build_parser().parse_args()

    try:
        failure_summary = read_input(args.input_file)
    except OSError as exc:
        exit_with_error(str(exc))

    try:
        from dv_agentic.agents.base import AgentConfig
        from dv_agentic.agents.bug_classifier import BugClassifierAgent

        llm = make_llm(model=args.model)
        agent = BugClassifierAgent(
            config=AgentConfig(name="bug_classifier", budget=args.budget),
            llm=llm,
            confidence_threshold=args.threshold,
        )
        result = asyncio.run(agent.run(failure_summary))
        print(result)  # noqa: T201
    except Exception as exc:
        exit_with_error(str(exc))

Coverage Analyst Command

coverage_analyst

CLI entrypoint for CoverageAnalystAgent.

Examples:

.. code-block:: shell

python3 -m dv_agentic.cli.coverage_analyst             --job-id my_test_42             --adapter imc             --threshold 90.0
main()

Main execution block of the CoverageAnalyst CLI.

Source code in src/dv_agentic/cli/coverage_analyst.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def main() -> None:
    """Main execution block of the CoverageAnalyst CLI."""
    args = _build_parser().parse_args()

    try:
        from dv_agentic.agents.base import AgentConfig
        from dv_agentic.agents.coverage_analyst import CoverageAnalystAgent
        from dv_agentic.tools.adapters import get_coverage_adapter

        coverage = get_coverage_adapter(args.adapter)
        agent = CoverageAnalystAgent(
            config=AgentConfig(name="coverage_analyst"),
            coverage=coverage,
            threshold=args.threshold,
        )
        result = asyncio.run(agent.run(args.job_id))
        print(result)  # noqa: T201
    except Exception as exc:
        exit_with_error(str(exc))

Reporter Command

reporter

CLI entrypoint for ReporterAgent.

Examples:

.. code-block:: shell

python3 -m dv_agentic.cli.reporter             --input-file session_results.txt             --output-path .agent/tasks/{task_id}_report.md
main()

Main execution block of the Reporter CLI.

Source code in src/dv_agentic/cli/reporter.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def main() -> None:
    """Main execution block of the Reporter CLI."""
    args = _build_parser().parse_args()

    try:
        session_results = read_input(args.input_file)
    except OSError as exc:
        exit_with_error(str(exc))

    output_path = args.output_path

    try:
        from dv_agentic.agents.base import AgentConfig
        from dv_agentic.agents.reporter import ReporterAgent

        llm = make_llm(model=args.model)
        agent = ReporterAgent(
            config=AgentConfig(name="reporter"),
            llm=llm,
            output_path=output_path,
        )
        result = asyncio.run(agent.run(session_results))
        print(result)  # noqa: T201
    except Exception as exc:
        exit_with_error(str(exc))

Environment Setup & Installer Command

install_agents

CLI entrypoint for the agent installer.

Materializes enriched .claude/agents/*.md (Claude Code YAML) and .opencode/agents/*.md (OpenCode YAML preserved from templates) from src/dv_agentic/prompts/*.tmpl.md, and mirrors root tools/ and skills/ into .claude/ / .opencode/.

What it does
  1. Optionally loads project.yaml + org profiles to enrich prompts (Level 1 injection: team rules + IP-type rules; session state omitted).
  2. For each of the agents, calls :class:~dv_agentic.prompts.prompt_loader.PromptLoader to produce an enriched prompt body (placeholders filled, unmatched removed).
  3. Strips the OpenCode-style YAML front matter from the source template.
  4. Prepends Claude Code compatible YAML front matter.
  5. Writes to {project_root}/.claude/agents/{agent}.md.
  6. Writes enriched content to {project_root}/.opencode/agents/{agent}.md, keeping the OpenCode YAML from the template.

Examples:

.. code-block:: shell

# No profile injection — raw prompts only
python3 -m dv_agentic.cli.install_agents --project-root /path/to/project

# Full profile injection
python3 -m dv_agentic.cli.install_agents \
    --project-root /path/to/project \
    --project-config .agent/project.yaml \
    --profiles-dir ../team-profiles

# Overwrite existing files
python3 -m dv_agentic.cli.install_agents --force
install(project_root, project_config_path=None, profiles_dir=None, force=False, target='opencode')

Standardized installer main entry.

Parameters:

Name Type Description Default
target str

Which platform(s) to install to — "claude", "opencode" (default), or "all".

 'opencode'
Source code in src/dv_agentic/cli/install_agents.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
def install(
    project_root: Path,
    project_config_path: str | None = None,
    profiles_dir: str | None = None,
    force: bool = False,
    target: str = "opencode",
) -> int:
    """Standardized installer main entry.

    Args:
        target: Which platform(s) to install to — ``"claude"``, ``"opencode"``
            (default), or ``"all"``.
    """
    project_ctx = _load_project_context(project_config_path, profiles_dir, project_root)

    from dv_agentic.prompts.prompt_loader import PromptLoader

    loader = PromptLoader(project_config=project_ctx)

    effective_targets = _merged_targets(target)
    _install_assets(
        project_root / "tools",
        "tools",
        project_root,
        force=force,
        targets_override=effective_targets,
    )
    _install_assets(
        project_root / "skills",
        "skills",
        project_root,
        force=force,
        targets_override=effective_targets,
    )

    # Agent directories: Claude Code and OpenCode each receive their own format.
    # .claude/agents/ → Claude Code YAML front matter
    # .opencode/agents/ → original OpenCode YAML front matter (from template)
    agent_writers: list[tuple[Path, Any]] = []
    if target in ("claude", "all"):
        claude_agent_dir = project_root / ".claude" / "agents"
        claude_agent_dir.mkdir(parents=True, exist_ok=True)
        agent_writers.append((claude_agent_dir, _write_agent_file))
    if target in ("opencode", "all"):
        opencode_agent_dir = project_root / ".opencode" / "agents"
        opencode_agent_dir.mkdir(parents=True, exist_ok=True)
        agent_writers.append((opencode_agent_dir, _write_opencode_agent_file))

    errors = 0
    written = 0
    skipped = 0

    for agent_name in _AGENTS:
        any_written = False
        any_error = False

        for agent_dir, write_fn in agent_writers:
            agent_md = agent_dir / f"{agent_name}.md"

            if agent_md.exists() and not force:
                logger.info("  skip  %s (exists)", agent_md.relative_to(project_root))
                skipped += 1
                continue

            try:
                write_fn(agent_name, loader, agent_md, project_root)
                any_written = True
            except FileNotFoundError:
                logger.warning("  warn %s — prompt template not found, skipping", agent_name)
                break
            except OSError as exc:
                logger.error("  ERROR writing %s: %s", agent_name, exc)
                any_error = True

        if any_written:
            written += 1
        if any_error:
            errors += 1

    print(  # noqa: T201
        f"\nInstall complete: {written} agents generated, {skipped} skipped, {errors} errors."
    )
    return 0 if errors == 0 else 1
main()

Main execution block for the install-agents CLI.

Source code in src/dv_agentic/cli/install_agents.py 
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
def main() -> None:
    """Main execution block for the install-agents CLI."""
    args = _build_parser().parse_args()

    # 1. Base path
    project_root = Path(args.project_root).resolve()

    # 2. Setup logging
    log_level = logging.DEBUG if args.verbose else logging.INFO
    logging.basicConfig(
        level=log_level,
        format="%(levelname)7s  %(message)s",
    )

    # 3. Execute
    rc = install(
        project_root=project_root,
        project_config_path=args.project_config,
        profiles_dir=args.profiles_dir,
        force=args.force,
        target=args.target,
    )
    sys.exit(rc)

Wiki Knowledge Base Commands

These commands manage the persistent LLM Wiki knowledge base introduced in v0.7.0. Wiki integration must be enabled via wiki.enabled: true in .agent/project.yaml.

Wiki Ingest Service

Ingest runs inside agents (LogAnalyzerAgent, BugClassifierAgent, ReporterAgent); there is no separate wiki_ingest CLI entry point.

ingest

Wiki ingest service — writes agent analysis results into the wiki.

Phase A scope

ingest_pattern() — create or update patterns/{failure_subtype}.md

Phase B scope (this file): ingest_bug() — create a new bugs/{TYPE}_{date}_{seq}.md page

Phase C will add: ingest_coverage_hole()

Design constraints
  • Every write is atomic (temp-file + os.replace).
  • log.md is append-only — existing entries are never modified.
  • Failures are non-fatal: callers catch all exceptions and log at DEBUG so a wiki write error never crashes an agent session.
  • No LLM client is required in Phase A/B — all content is structured data derived from LogAnalyzerAgent / BugClassifierAgent output.
WikiIngestResult dataclass

Summary of a completed :class:WikiIngestService operation.

Attributes:

Name Type Description
task_id str

Originating task identifier (used for citation).
pages_created list[str]

Relative paths of newly created wiki pages.
pages_updated list[str]

Relative paths of updated wiki pages.
log_entry str

The text appended to log.md.
index_updated bool

Whether index.md was regenerated.
search_index_updated bool

Whether the BM25 index was refreshed.
Source code in src/dv_agentic/wiki/ingest.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@dataclass
class WikiIngestResult:
    """Summary of a completed :class:`WikiIngestService` operation.

    Attributes:
        task_id: Originating task identifier (used for citation).
        pages_created: Relative paths of newly created wiki pages.
        pages_updated: Relative paths of updated wiki pages.
        log_entry: The text appended to ``log.md``.
        index_updated: Whether ``index.md`` was regenerated.
        search_index_updated: Whether the BM25 index was refreshed.
    """

    task_id: str
    pages_created: list[str] = field(default_factory=list)
    pages_updated: list[str] = field(default_factory=list)
    log_entry: str = ""
    index_updated: bool = False
    search_index_updated: bool = False
WikiIngestService

Writes LogAnalyzerAgent results into the wiki knowledge base.

Parameters:

Name Type Description Default
wiki_config WikiConfig

Wiki integration configuration.

 required
Source code in src/dv_agentic/wiki/ingest.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
class WikiIngestService:
    """Writes ``LogAnalyzerAgent`` results into the wiki knowledge base.

    Args:
        wiki_config: Wiki integration configuration.
    """

    def __init__(self, wiki_config: WikiConfig) -> None:
        self.cfg = wiki_config
        self._search_index: Any | None = None  # lazy-initialised

    # ------------------------------------------------------------------
    # Phase A public API
    # ------------------------------------------------------------------

    def ingest_pattern(
        self,
        failure_subtype: str,
        error_class: str,
        context_lines: list[str],
        fix_applied: str | None,
        success: bool,
        task_id: str,
    ) -> WikiIngestResult:
        """Create or update ``patterns/{failure_subtype}.md``.

        On first occurrence the page is created from a template.  On
        subsequent calls only ``hit_count``, ``last_seen``, and the
        Resolution History table are updated.

        Args:
            failure_subtype: Granular subtype from
                :meth:`~dv_agentic.agents.log_analyzer.LogAnalyzerAgent._classify_subtype`
                (e.g. ``"missing_timescale"``).
            error_class: Top-level error class (e.g. ``"compile_error"``).
            context_lines: Lines surrounding the matched error in the log.
            fix_applied: Short description of the fix applied, or ``None``.
            success: Whether the fix led to a passing simulation.
            task_id: Task identifier used in the citation footer.

        Returns:
            :class:`WikiIngestResult` describing what was written.
        """
        result = WikiIngestResult(task_id=task_id)
        patterns_dir = self.cfg.wiki_dir / "patterns"
        patterns_dir.mkdir(parents=True, exist_ok=True)

        page_path = patterns_dir / f"{failure_subtype}.md"
        today = today_str()

        if page_path.exists():
            result = self._update_pattern_page(
                page_path, fix_applied, success, task_id, today, result
            )
        else:
            result = self._create_pattern_page(
                page_path,
                failure_subtype,
                error_class,
                context_lines,
                fix_applied,
                success,
                task_id,
                today,
                result,
            )

        # Refresh search index (non-fatal)
        try:
            idx = self._get_search_index()
            idx.update(page_path, page_path.read_text(encoding="utf-8"))
            result.search_index_updated = True
        except Exception:
            logger.debug("Wiki: search index update failed (non-fatal)", exc_info=True)

        # Append to log.md
        log_entry = self._build_log_entry(
            task_id, error_class, failure_subtype, fix_applied, success, result
        )
        result.log_entry = log_entry
        self._append_log(log_entry)

        # Regenerate patterns section of index.md
        self._update_index_patterns(patterns_dir)
        result.index_updated = True

        return result

    # ------------------------------------------------------------------
    # Phase B public API
    # ------------------------------------------------------------------

    def ingest_bug(
        self,
        bug_type: str,
        confidence: float,
        evidence: list[str],
        error_class: str,
        failure_subtype: str,
        task_id: str,
        ip_type: str = "custom",
        log_path: str = "",
    ) -> WikiIngestResult:
        """Create a new bug page in ``bugs/{TYPE}_{date}_{seq:03d}.md``.

        Every call creates a *new* page (bugs are not merged on update).
        The page ID follows: ``{RTL|TB}_{YYYYMMDD}_{seq:03d}``.

        Args:
            bug_type: ``"RTL_BUG"`` or ``"TB_BUG"``.
            confidence: Classification confidence score (0.0-1.0).
            evidence: Bullet-point evidence strings from BugClassifierAgent.
            error_class: Top-level error class from LogAnalyzerAgent.
            failure_subtype: Granular subtype from LogAnalyzerAgent.
            task_id: Task identifier used in citation and log.
            ip_type: IP type (``"axi"``, ``"pcie"``, ``"ddr"``, ``"custom"``).
            log_path: Path to the simulation log (for citation).

        Returns:
            :class:`WikiIngestResult` describing what was written.
        """
        result = WikiIngestResult(task_id=task_id)
        bugs_dir = self.cfg.wiki_dir / "bugs"
        bugs_dir.mkdir(parents=True, exist_ok=True)

        today = today_str()
        prefix = "RTL" if bug_type == "RTL_BUG" else "TB"
        date_str = today.replace("-", "")
        seq = self._next_bug_seq(bugs_dir, prefix, date_str)
        bug_id = f"{prefix}_{date_str}_{seq:03d}"
        page_path = bugs_dir / f"{bug_id}.md"

        frontmatter: dict[str, Any] = {
            "id": bug_id,
            "type": bug_type,
            "status": "open",
            "confidence": round(confidence, 4),
            "first_seen": today,
            "last_updated": today,
            "task_ids": [task_id],
            "error_class": error_class,
            "failure_subtype": failure_subtype,
            "ip_type": ip_type,
        }

        evidence_lines = "\n".join(f"- {e}" for e in evidence) if evidence else "- (none)"
        log_ref = f"(source: log: {log_path})" if log_path else ""
        body = _BUG_BODY_TEMPLATE.format(
            bug_id=bug_id,
            bug_type=bug_type,
            evidence_lines=evidence_lines,
            log_ref=log_ref,
        )
        atomic_write(page_path, serialize_page(frontmatter, body))
        rel = str(page_path.relative_to(self.cfg.wiki_dir))
        result.pages_created.append(rel)
        logger.info("Wiki: created bug page %s", bug_id)

        # Refresh search index (non-fatal)
        try:
            idx = self._get_search_index()
            idx.update(page_path, page_path.read_text(encoding="utf-8"))
            result.search_index_updated = True
        except Exception:
            logger.debug("Wiki: search index update failed (non-fatal)", exc_info=True)

        # Append to log.md
        log_entry = self._build_bug_log_entry(bug_id, bug_type, task_id, result)
        result.log_entry = log_entry
        self._append_log(log_entry)

        # Regenerate bugs section of index.md
        self._update_index_bugs(bugs_dir)
        result.index_updated = True

        return result

    # ------------------------------------------------------------------
    # Phase C public API
    # ------------------------------------------------------------------

    def ingest_coverage_hole(
        self,
        covergroup: str,
        bin_name: str,
        action_class: str,
        scenario: str,
        filled: bool,
        task_id: str,
    ) -> WikiIngestResult:
        """Create or update coverage/{covergroup}_{bin_name}.md."""
        result = WikiIngestResult(task_id=task_id)
        cov_dir = self.cfg.wiki_dir / "coverage"
        cov_dir.mkdir(parents=True, exist_ok=True)

        page_id = f"{covergroup}_{bin_name}"
        page_path = cov_dir / f"{page_id}.md"
        today = today_str()

        frontmatter: dict[str, Any] = {
            "type": "COVERAGE_HOLE",
            "covergroup": covergroup,
            "bin": bin_name,
            "action_class": action_class,
            "filled": filled,
            "last_updated": today,
            "task_id": task_id,
        }
        body = f"## Scenario\n{scenario}\n"

        if page_path.exists():
            old_fm, old_body = parse_page(page_path.read_text(encoding="utf-8"))
            frontmatter["first_seen"] = old_fm.get("first_seen", today)
            body = old_body.rstrip() + f"\n\n## Update [{today}] (Task: {task_id})\n{scenario}\n"
            result.pages_updated.append(f"coverage/{page_path.name}")
        else:
            frontmatter["first_seen"] = today
            result.pages_created.append(f"coverage/{page_path.name}")

        atomic_write(page_path, serialize_page(frontmatter, body))
        logger.info("Wiki: ingested coverage hole %s", page_path.name)

        # Search index
        try:
            idx = self._get_search_index()
            idx.update(page_path, page_path.read_text(encoding="utf-8"))
            result.search_index_updated = True
        except Exception:
            logger.debug("Wiki: search index update failed (non-fatal)")

        # log.md
        log_entry = self._build_coverage_log_entry(page_id, task_id, result)
        result.log_entry = log_entry
        self._append_log(log_entry)

        # index.md
        self._update_index_coverage(cov_dir)
        result.index_updated = True

        return result

    def ingest_session(
        self,
        session_report: str,
        failure_summary: str | None,
        classification: str | None,
        coverage_summary: str | None,
        task_id: str,
    ) -> WikiIngestResult:
        """Heuristically dispatch to pattern, bug, and coverage ingestion based on session data."""
        result = WikiIngestResult(task_id=task_id)

        error_class = "unknown"
        failure_subtype = "unknown"

        # Pattern parsing
        if failure_summary:
            for line in failure_summary.splitlines():
                if line.startswith("error_class:"):
                    error_class = line.split(":", 1)[1].strip()
                elif line.startswith("failure_subtype:"):
                    failure_subtype = line.split(":", 1)[1].strip()

            if failure_subtype != "unknown":
                p_res = self.ingest_pattern(
                    failure_subtype=failure_subtype,
                    error_class=error_class,
                    context_lines=[],
                    fix_applied=None,
                    success=False,
                    task_id=task_id,
                )
                result.pages_created.extend(p_res.pages_created)
                result.pages_updated.extend(p_res.pages_updated)

        # Bug parsing
        if classification and "BUG_TYPE" in classification:
            bug_type = "UNKNOWN"
            confidence = 0.0
            import re

            m_type = re.search(r"BUG_TYPE\s*:\s*(\w+)", classification)
            if m_type:
                bug_type = m_type.group(1)
            m_conf = re.search(r"CONFIDENCE\s*:\s*([0-9\.]+)", classification)
            if m_conf:
                import contextlib

                with contextlib.suppress(ValueError):
                    confidence = float(m_conf.group(1))
            if bug_type != "UNKNOWN":
                b_res = self.ingest_bug(
                    bug_type=bug_type,
                    confidence=confidence,
                    evidence=["Extracted from session"],
                    error_class=error_class,
                    failure_subtype=failure_subtype,
                    task_id=task_id,
                )
                result.pages_created.extend(b_res.pages_created)
                result.pages_updated.extend(b_res.pages_updated)

        # Append session report to log
        entry = f"## [{today_str()}] SESSION_REPORT | {task_id}\n\n{session_report}"
        self._append_log(entry)
        result.log_entry = entry

        return result

    # ------------------------------------------------------------------
    # Private — page creation / update
    # ------------------------------------------------------------------

    def _create_pattern_page(
        self,
        page_path: Path,
        pattern_id: str,
        error_class: str,
        context_lines: list[str],
        fix_applied: str | None,
        success: bool,
        task_id: str,
        today: str,
        result: WikiIngestResult,
    ) -> WikiIngestResult:
        """Write a new pattern page from the template."""
        fix_str = fix_applied or "—"
        fix_result = "PASS" if success else "—"
        first_row = f"| {today} | {task_id} | {fix_str} | {fix_result} |"
        signature = "\n".join(context_lines[:3]) if context_lines else "(see log)"

        frontmatter: dict[str, Any] = {
            "pattern_id": pattern_id,
            "error_class": error_class,
            "failure_subtype": pattern_id,
            "hit_count": 1,
            "first_seen": today,
            "last_seen": today,
            "fix_success_rate": None,
            "_success_count": 1 if (fix_applied and success) else 0,
        }
        body = _PATTERN_BODY_TEMPLATE.format(
            pattern_id=pattern_id,
            signature=signature,
            first_row=first_row,
        )
        atomic_write(page_path, serialize_page(frontmatter, body))
        rel = str(page_path.relative_to(self.cfg.wiki_dir))
        result.pages_created.append(rel)
        logger.info("Wiki: created pattern page %s", page_path.name)
        return result

    def _update_pattern_page(
        self,
        page_path: Path,
        fix_applied: str | None,
        success: bool,
        task_id: str,
        today: str,
        result: WikiIngestResult,
    ) -> WikiIngestResult:
        """Increment ``hit_count``, update ``last_seen``, append history row."""
        content = page_path.read_text(encoding="utf-8")
        frontmatter, body = parse_page(content)

        # --- update counters ---
        hit_count: int = int(frontmatter.get("hit_count", 0)) + 1
        frontmatter["hit_count"] = hit_count
        frontmatter["last_seen"] = today

        # Recalculate fix_success_rate when a fix was actually provided
        if fix_applied is not None:
            successes: int = int(frontmatter.get("_success_count", 0)) + (1 if success else 0)
            frontmatter["_success_count"] = successes
            frontmatter["fix_success_rate"] = round(successes / hit_count, 2)

        # --- append resolution history row ---
        fix_str = fix_applied or "—"
        fix_result = "PASS" if success else "—"
        new_row = f"| {today} | {task_id} | {fix_str} | {fix_result} |"
        body = body.rstrip() + f"\n{new_row}\n"

        atomic_write(page_path, serialize_page(frontmatter, body))
        rel = str(page_path.relative_to(self.cfg.wiki_dir))
        result.pages_updated.append(rel)
        logger.info("Wiki: updated pattern page %s (hit_count=%d)", page_path.name, hit_count)
        return result

    # ------------------------------------------------------------------
    # Private — log.md
    # ------------------------------------------------------------------

    @staticmethod
    def _build_log_entry(
        task_id: str,
        error_class: str,
        failure_subtype: str,
        fix_applied: str | None,
        success: bool,
        result: WikiIngestResult,
    ) -> str:
        today = today_str()
        fix_str = fix_applied or "—"
        lines = [
            f"## [{today}] ingest_pattern | {task_id} | {failure_subtype}",
            f"- error_class: {error_class}",
            f"- failure_subtype: {failure_subtype}",
            f"- fix_applied: {fix_str}",
            f"- success: {success}",
            "- wiki_pages_updated:",
        ]
        for p in result.pages_created + result.pages_updated:
            lines.append(f"    - {p}")
        return "\n".join(lines)

    def _append_log(self, entry: str) -> None:
        """Append *entry* to ``log.md``.  **Never** truncates or modifies
        existing content — ``log.md`` is strictly append-only.
        """
        log_path = self.cfg.wiki_dir / "log.md"
        log_path.parent.mkdir(parents=True, exist_ok=True)

        if not log_path.exists():
            header = "# DV Agentic Wiki Operation Log\n\n"
            atomic_write(log_path, header + entry + "\n")
        else:
            with log_path.open("a", encoding="utf-8") as fh:
                fh.write("\n" + entry + "\n")

    # ------------------------------------------------------------------
    # Private — index.md regeneration
    # ------------------------------------------------------------------

    def _update_index_patterns(self, patterns_dir: Path) -> None:
        """Regenerate the Patterns table inside ``wiki/index.md``."""
        index_path = self.cfg.wiki_dir / "index.md"
        rows: list[str] = []

        for page in sorted(patterns_dir.glob("*.md")):
            if page.name.startswith("_"):
                continue
            try:
                fm, _ = parse_page(page.read_text(encoding="utf-8"))
                pid = fm.get("pattern_id", page.stem)
                hits = fm.get("hit_count", 0)
                rate = fm.get("fix_success_rate")
                rate_str = f"{rate:.0%}" if isinstance(rate, int | float) else "N/A"
                link = f"patterns/{page.name}"
                rows.append(f"| [{pid}]({link}) | {pid} | {hits} | {rate_str} |")
            except Exception:
                logger.debug("Wiki: could not parse pattern page %s", page)

        section = (
            f"## Patterns ({len(rows)} pages)\n"
            "| Page | failure_subtype | hit_count | fix_success_rate |\n"
            "|------|----------------|-----------|------------------|\n"
        )
        section += ("\n".join(rows) if rows else "| — | — | — | — |") + "\n"

        if not index_path.exists():
            content = (
                "# DV Agentic Wiki Index\n\n"
                f"Last updated: {now_iso()} | Total pages: {len(rows)}\n\n" + section
            )
            atomic_write(index_path, content)
            return

        existing = index_path.read_text(encoding="utf-8")
        if "## Patterns" in existing:
            before = existing[: existing.index("## Patterns")]
            rest = existing[existing.index("## Patterns") :]
            next_h2 = rest.find("\n## ", 4)
            after = rest[next_h2:] if next_h2 != -1 else ""
            new_content = before + section + after
        else:
            new_content = existing.rstrip() + "\n\n" + section

        atomic_write(index_path, new_content)

    # ------------------------------------------------------------------
    # Private — search index (lazy init)
    # ------------------------------------------------------------------

    def _get_search_index(self) -> Any:
        if self._search_index is None:
            from .search import WikiSearchIndex

            self._search_index = WikiSearchIndex.create(self.cfg.search_backend, self.cfg.wiki_dir)
            if self.cfg.wiki_dir.is_dir():
                self._search_index.build(self.cfg.wiki_dir)
        return self._search_index

    # ------------------------------------------------------------------
    # Private — Phase B helpers
    # ------------------------------------------------------------------

    @staticmethod
    def _next_bug_seq(bugs_dir: Path, prefix: str, date_str: str) -> int:
        """Return the next available sequence number for today's bug pages.

        Scans existing ``{PREFIX}_{date_str}_NNN.md`` files and returns
        ``max_seq + 1``, or ``1`` if none exist.
        """
        import re

        pattern = re.compile(rf"^{re.escape(prefix)}_{re.escape(date_str)}_(\d{{3}})\.md$")
        max_seq = 0
        for p in bugs_dir.glob(f"{prefix}_{date_str}_*.md"):
            m = pattern.match(p.name)
            if m:
                max_seq = max(max_seq, int(m.group(1)))
        return max_seq + 1

    @staticmethod
    def _build_bug_log_entry(
        bug_id: str,
        bug_type: str,
        task_id: str,
        result: WikiIngestResult,
    ) -> str:
        today = today_str()
        lines = [
            f"## [{today}] ingest_bug | {task_id} | {bug_id}",
            f"- type: {bug_type}",
            f"- task_id: {task_id}",
            "- wiki_pages_updated:",
        ]
        for p in result.pages_created + result.pages_updated:
            lines.append(f"    - {p}")
        return "\n".join(lines)

    def _update_index_bugs(self, bugs_dir: Path) -> None:
        """Regenerate the Bugs table inside ``wiki/index.md``."""
        index_path = self.cfg.wiki_dir / "index.md"
        rows: list[str] = []

        for page in sorted(bugs_dir.glob("*.md")):
            if page.name.startswith("_"):
                continue
            try:
                fm, _ = parse_page(page.read_text(encoding="utf-8"))
                bug_id = fm.get("id", page.stem)
                btype = fm.get("type", "?")
                status = fm.get("status", "?")
                conf = fm.get("confidence", "?")
                last_upd = fm.get("last_updated", "?")
                link = f"bugs/{page.name}"
                rows.append(f"| [{bug_id}]({link}) | {btype} | {status} | {conf} | {last_upd} |")
            except Exception:
                logger.debug("Wiki: could not parse bug page %s", page)

        section = (
            f"## Bugs ({len(rows)} pages)\n"
            "| Page | Type | Status | Confidence | Last Updated |\n"
            "|------|------|--------|------------|--------------|\n"
        )
        section += ("\n".join(rows) if rows else "| — | — | — | — | — |") + "\n"

        if not index_path.exists():
            content = (
                "# DV Agentic Wiki Index\n\n"
                f"Last updated: {now_iso()} | Total pages: {len(rows)}\n\n" + section
            )
            atomic_write(index_path, content)
            return

        existing = index_path.read_text(encoding="utf-8")
        if "## Bugs" in existing:
            before = existing[: existing.index("## Bugs")]
            rest = existing[existing.index("## Bugs") :]
            next_h2 = rest.find("\n## ", 4)
            after = rest[next_h2:] if next_h2 != -1 else ""
            new_content = before + section + after
        else:
            new_content = existing.rstrip() + "\n\n" + section

        atomic_write(index_path, new_content)

    @staticmethod
    def _build_coverage_log_entry(
        page_id: str,
        task_id: str,
        result: WikiIngestResult,
    ) -> str:
        today = today_str()
        lines = [
            f"## [{today}] ingest_coverage_hole | {task_id} | {page_id}",
            f"- task_id: {task_id}",
            "- wiki_pages_updated:",
        ]
        for p in result.pages_created + result.pages_updated:
            lines.append(f"    - {p}")
        return "\n".join(lines)

    def _update_index_coverage(self, cov_dir: Path) -> None:
        """Regenerate the Coverage table inside ``wiki/index.md``."""
        index_path = self.cfg.wiki_dir / "index.md"
        rows: list[str] = []

        for page in sorted(cov_dir.glob("*.md")):
            if page.name.startswith("_"):
                continue
            try:
                fm, _ = parse_page(page.read_text(encoding="utf-8"))
                cg = fm.get("covergroup", "?")
                b_name = fm.get("bin", "?")
                acls = fm.get("action_class", "?")
                filled = fm.get("filled", "?")
                link = f"coverage/{page.name}"
                rows.append(f"| [{cg}_{b_name}]({link}) | {cg} | {b_name} | {acls} | {filled} |")
            except Exception:
                logger.debug("Wiki: could not parse coverage page %s", page)

        section = (
            f"## Coverage Holes ({len(rows)} pages)\n"
            "| Page | Covergroup | Bin | Action Class | Filled |\n"
            "|------|------------|-----|--------------|--------|\n"
        )
        section += ("\n".join(rows) if rows else "| — | — | — | — | — |") + "\n"

        if not index_path.exists():
            content = (
                "# DV Agentic Wiki Index\n\n"
                f"Last updated: {now_iso()} | Total pages: {len(rows)}\n\n" + section
            )
            atomic_write(index_path, content)
            return

        existing = index_path.read_text(encoding="utf-8")
        if "## Coverage Holes" in existing:
            before = existing[: existing.index("## Coverage Holes")]
            rest = existing[existing.index("## Coverage Holes") :]
            next_h2 = rest.find("\n## ", 4)
            after = rest[next_h2:] if next_h2 != -1 else ""
            new_content = before + section + after
        else:
            new_content = existing.rstrip() + "\n\n" + section

        atomic_write(index_path, new_content)
ingest_bug(bug_type, confidence, evidence, error_class, failure_subtype, task_id, ip_type='custom', log_path='')

Create a new bug page in bugs/{TYPE}_{date}_{seq:03d}.md.

Every call creates a new page (bugs are not merged on update). The page ID follows: {RTL|TB}_{YYYYMMDD}_{seq:03d}.

Parameters:

Name Type Description Default
bug_type str

"RTL_BUG" or "TB_BUG".

 required
confidence float

Classification confidence score (0.0-1.0).

 required
evidence list[str]

Bullet-point evidence strings from BugClassifierAgent.

 required
error_class str

Top-level error class from LogAnalyzerAgent.

 required
failure_subtype str

Granular subtype from LogAnalyzerAgent.

 required
task_id str

Task identifier used in citation and log.

 required
ip_type str

IP type ("axi", "pcie", "ddr", "custom").

 'custom'
log_path str

Path to the simulation log (for citation).

 ''

Returns:

Type Description
WikiIngestResult

class:WikiIngestResult describing what was written.
Source code in src/dv_agentic/wiki/ingest.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def ingest_bug(
    self,
    bug_type: str,
    confidence: float,
    evidence: list[str],
    error_class: str,
    failure_subtype: str,
    task_id: str,
    ip_type: str = "custom",
    log_path: str = "",
) -> WikiIngestResult:
    """Create a new bug page in ``bugs/{TYPE}_{date}_{seq:03d}.md``.

    Every call creates a *new* page (bugs are not merged on update).
    The page ID follows: ``{RTL|TB}_{YYYYMMDD}_{seq:03d}``.

    Args:
        bug_type: ``"RTL_BUG"`` or ``"TB_BUG"``.
        confidence: Classification confidence score (0.0-1.0).
        evidence: Bullet-point evidence strings from BugClassifierAgent.
        error_class: Top-level error class from LogAnalyzerAgent.
        failure_subtype: Granular subtype from LogAnalyzerAgent.
        task_id: Task identifier used in citation and log.
        ip_type: IP type (``"axi"``, ``"pcie"``, ``"ddr"``, ``"custom"``).
        log_path: Path to the simulation log (for citation).

    Returns:
        :class:`WikiIngestResult` describing what was written.
    """
    result = WikiIngestResult(task_id=task_id)
    bugs_dir = self.cfg.wiki_dir / "bugs"
    bugs_dir.mkdir(parents=True, exist_ok=True)

    today = today_str()
    prefix = "RTL" if bug_type == "RTL_BUG" else "TB"
    date_str = today.replace("-", "")
    seq = self._next_bug_seq(bugs_dir, prefix, date_str)
    bug_id = f"{prefix}_{date_str}_{seq:03d}"
    page_path = bugs_dir / f"{bug_id}.md"

    frontmatter: dict[str, Any] = {
        "id": bug_id,
        "type": bug_type,
        "status": "open",
        "confidence": round(confidence, 4),
        "first_seen": today,
        "last_updated": today,
        "task_ids": [task_id],
        "error_class": error_class,
        "failure_subtype": failure_subtype,
        "ip_type": ip_type,
    }

    evidence_lines = "\n".join(f"- {e}" for e in evidence) if evidence else "- (none)"
    log_ref = f"(source: log: {log_path})" if log_path else ""
    body = _BUG_BODY_TEMPLATE.format(
        bug_id=bug_id,
        bug_type=bug_type,
        evidence_lines=evidence_lines,
        log_ref=log_ref,
    )
    atomic_write(page_path, serialize_page(frontmatter, body))
    rel = str(page_path.relative_to(self.cfg.wiki_dir))
    result.pages_created.append(rel)
    logger.info("Wiki: created bug page %s", bug_id)

    # Refresh search index (non-fatal)
    try:
        idx = self._get_search_index()
        idx.update(page_path, page_path.read_text(encoding="utf-8"))
        result.search_index_updated = True
    except Exception:
        logger.debug("Wiki: search index update failed (non-fatal)", exc_info=True)

    # Append to log.md
    log_entry = self._build_bug_log_entry(bug_id, bug_type, task_id, result)
    result.log_entry = log_entry
    self._append_log(log_entry)

    # Regenerate bugs section of index.md
    self._update_index_bugs(bugs_dir)
    result.index_updated = True

    return result
ingest_coverage_hole(covergroup, bin_name, action_class, scenario, filled, task_id)

Create or update coverage/{covergroup}_{bin_name}.md.

Source code in src/dv_agentic/wiki/ingest.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
def ingest_coverage_hole(
    self,
    covergroup: str,
    bin_name: str,
    action_class: str,
    scenario: str,
    filled: bool,
    task_id: str,
) -> WikiIngestResult:
    """Create or update coverage/{covergroup}_{bin_name}.md."""
    result = WikiIngestResult(task_id=task_id)
    cov_dir = self.cfg.wiki_dir / "coverage"
    cov_dir.mkdir(parents=True, exist_ok=True)

    page_id = f"{covergroup}_{bin_name}"
    page_path = cov_dir / f"{page_id}.md"
    today = today_str()

    frontmatter: dict[str, Any] = {
        "type": "COVERAGE_HOLE",
        "covergroup": covergroup,
        "bin": bin_name,
        "action_class": action_class,
        "filled": filled,
        "last_updated": today,
        "task_id": task_id,
    }
    body = f"## Scenario\n{scenario}\n"

    if page_path.exists():
        old_fm, old_body = parse_page(page_path.read_text(encoding="utf-8"))
        frontmatter["first_seen"] = old_fm.get("first_seen", today)
        body = old_body.rstrip() + f"\n\n## Update [{today}] (Task: {task_id})\n{scenario}\n"
        result.pages_updated.append(f"coverage/{page_path.name}")
    else:
        frontmatter["first_seen"] = today
        result.pages_created.append(f"coverage/{page_path.name}")

    atomic_write(page_path, serialize_page(frontmatter, body))
    logger.info("Wiki: ingested coverage hole %s", page_path.name)

    # Search index
    try:
        idx = self._get_search_index()
        idx.update(page_path, page_path.read_text(encoding="utf-8"))
        result.search_index_updated = True
    except Exception:
        logger.debug("Wiki: search index update failed (non-fatal)")

    # log.md
    log_entry = self._build_coverage_log_entry(page_id, task_id, result)
    result.log_entry = log_entry
    self._append_log(log_entry)

    # index.md
    self._update_index_coverage(cov_dir)
    result.index_updated = True

    return result
ingest_pattern(failure_subtype, error_class, context_lines, fix_applied, success, task_id)

Create or update patterns/{failure_subtype}.md.

On first occurrence the page is created from a template. On subsequent calls only hit_count, last_seen, and the Resolution History table are updated.

Parameters:

Name Type Description Default
failure_subtype str

Granular subtype from :meth:~dv_agentic.agents.log_analyzer.LogAnalyzerAgent._classify_subtype (e.g. "missing_timescale").

 required
error_class str

Top-level error class (e.g. "compile_error").

 required
context_lines list[str]

Lines surrounding the matched error in the log.

 required
fix_applied str | None

Short description of the fix applied, or None.

 required
success bool

Whether the fix led to a passing simulation.

 required
task_id str

Task identifier used in the citation footer.

 required

Returns:

Type Description
WikiIngestResult

class:WikiIngestResult describing what was written.
Source code in src/dv_agentic/wiki/ingest.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def ingest_pattern(
    self,
    failure_subtype: str,
    error_class: str,
    context_lines: list[str],
    fix_applied: str | None,
    success: bool,
    task_id: str,
) -> WikiIngestResult:
    """Create or update ``patterns/{failure_subtype}.md``.

    On first occurrence the page is created from a template.  On
    subsequent calls only ``hit_count``, ``last_seen``, and the
    Resolution History table are updated.

    Args:
        failure_subtype: Granular subtype from
            :meth:`~dv_agentic.agents.log_analyzer.LogAnalyzerAgent._classify_subtype`
            (e.g. ``"missing_timescale"``).
        error_class: Top-level error class (e.g. ``"compile_error"``).
        context_lines: Lines surrounding the matched error in the log.
        fix_applied: Short description of the fix applied, or ``None``.
        success: Whether the fix led to a passing simulation.
        task_id: Task identifier used in the citation footer.

    Returns:
        :class:`WikiIngestResult` describing what was written.
    """
    result = WikiIngestResult(task_id=task_id)
    patterns_dir = self.cfg.wiki_dir / "patterns"
    patterns_dir.mkdir(parents=True, exist_ok=True)

    page_path = patterns_dir / f"{failure_subtype}.md"
    today = today_str()

    if page_path.exists():
        result = self._update_pattern_page(
            page_path, fix_applied, success, task_id, today, result
        )
    else:
        result = self._create_pattern_page(
            page_path,
            failure_subtype,
            error_class,
            context_lines,
            fix_applied,
            success,
            task_id,
            today,
            result,
        )

    # Refresh search index (non-fatal)
    try:
        idx = self._get_search_index()
        idx.update(page_path, page_path.read_text(encoding="utf-8"))
        result.search_index_updated = True
    except Exception:
        logger.debug("Wiki: search index update failed (non-fatal)", exc_info=True)

    # Append to log.md
    log_entry = self._build_log_entry(
        task_id, error_class, failure_subtype, fix_applied, success, result
    )
    result.log_entry = log_entry
    self._append_log(log_entry)

    # Regenerate patterns section of index.md
    self._update_index_patterns(patterns_dir)
    result.index_updated = True

    return result
ingest_session(session_report, failure_summary, classification, coverage_summary, task_id)

Heuristically dispatch to pattern, bug, and coverage ingestion based on session data.

Source code in src/dv_agentic/wiki/ingest.py 
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
def ingest_session(
    self,
    session_report: str,
    failure_summary: str | None,
    classification: str | None,
    coverage_summary: str | None,
    task_id: str,
) -> WikiIngestResult:
    """Heuristically dispatch to pattern, bug, and coverage ingestion based on session data."""
    result = WikiIngestResult(task_id=task_id)

    error_class = "unknown"
    failure_subtype = "unknown"

    # Pattern parsing
    if failure_summary:
        for line in failure_summary.splitlines():
            if line.startswith("error_class:"):
                error_class = line.split(":", 1)[1].strip()
            elif line.startswith("failure_subtype:"):
                failure_subtype = line.split(":", 1)[1].strip()

        if failure_subtype != "unknown":
            p_res = self.ingest_pattern(
                failure_subtype=failure_subtype,
                error_class=error_class,
                context_lines=[],
                fix_applied=None,
                success=False,
                task_id=task_id,
            )
            result.pages_created.extend(p_res.pages_created)
            result.pages_updated.extend(p_res.pages_updated)

    # Bug parsing
    if classification and "BUG_TYPE" in classification:
        bug_type = "UNKNOWN"
        confidence = 0.0
        import re

        m_type = re.search(r"BUG_TYPE\s*:\s*(\w+)", classification)
        if m_type:
            bug_type = m_type.group(1)
        m_conf = re.search(r"CONFIDENCE\s*:\s*([0-9\.]+)", classification)
        if m_conf:
            import contextlib

            with contextlib.suppress(ValueError):
                confidence = float(m_conf.group(1))
        if bug_type != "UNKNOWN":
            b_res = self.ingest_bug(
                bug_type=bug_type,
                confidence=confidence,
                evidence=["Extracted from session"],
                error_class=error_class,
                failure_subtype=failure_subtype,
                task_id=task_id,
            )
            result.pages_created.extend(b_res.pages_created)
            result.pages_updated.extend(b_res.pages_updated)

    # Append session report to log
    entry = f"## [{today_str()}] SESSION_REPORT | {task_id}\n\n{session_report}"
    self._append_log(entry)
    result.log_entry = entry

    return result

Wiki Lint Command

wiki_lint

Wiki linting CLI.

Usage

python -m dv_agentic.cli.wiki_lint [--depth quick|full]

Wiki Search Index

search

Wiki search index — BM25 and keyword fallback backends.

Strategy Pattern: two implementations share one WikiSearchIndex interface.

BM25SearchIndex — uses bm25s[core] (air-gapped, pure Python). Falls back to KeywordSearchIndex when bm25s is absent. KeywordSearchIndex — simple substring scoring; zero extra dependencies.

Usage::

idx = WikiSearchIndex.create("bm25", wiki_dir)
idx.build(wiki_dir)                         # initial build
idx.update(page_path, new_content)          # after each ingest write
results = idx.search("timescale", top_k=3)  # BM25 / keyword query

Install bm25s for the full BM25 experience::

pip install "bm25s[core]"

Without it, the system transparently falls back to keyword matching so every agent workflow keeps working.

BM25SearchIndex

Bases: WikiSearchIndex

BM25 index backed by bm25s[core].

Falls back to :class:KeywordSearchIndex transparently when bm25s is not installed, so the system keeps working in environments where the [wiki] optional extra is absent.

The bm25s index is persisted to {wiki_dir}/.search_index/ after each full build. Because bm25s v0.2 does not support incremental updates, a full rebuild is triggered on every :meth:update call. For wiki sizes typical in DV projects (< 500 pages) this takes < 1 s.

Source code in src/dv_agentic/wiki/search.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
class BM25SearchIndex(WikiSearchIndex):
    """BM25 index backed by ``bm25s[core]``.

    Falls back to :class:`KeywordSearchIndex` transparently when ``bm25s``
    is not installed, so the system keeps working in environments where the
    ``[wiki]`` optional extra is absent.

    The bm25s index is **persisted** to ``{wiki_dir}/.search_index/`` after
    each full build.  Because ``bm25s`` v0.2 does not support incremental
    updates, a full rebuild is triggered on every :meth:`update` call.
    For wiki sizes typical in DV projects (< 500 pages) this takes < 1 s.
    """

    def __init__(self, wiki_dir: Path) -> None:
        self._wiki_dir = wiki_dir
        self._index_path = wiki_dir / _INDEX_SUBDIR
        # In-memory page cache used to avoid re-reading during search
        self._pages: list[tuple[Path, str]] = []
        # Set when bm25s is unavailable
        self._fallback: KeywordSearchIndex | None = None

    # ------------------------------------------------------------------
    # Public interface
    # ------------------------------------------------------------------

    def build(self, wiki_dir: Path) -> None:
        """Build BM25 index from all wiki pages, persist to disk."""
        try:
            self._build_bm25(wiki_dir)
        except ImportError:
            logger.info(
                "bm25s not installed — wiki search uses keyword fallback. "
                "Install with: pip install 'bm25s[core]'"
            )
            self._fallback = KeywordSearchIndex(wiki_dir)
            self._fallback.build(wiki_dir)

    def update(self, page_path: Path, content: str) -> None:
        """Rebuild the full index after a page is written."""
        if self._fallback is not None:
            self._fallback.update(page_path, content)
            return
        # bm25s v0.2: no incremental API — full rebuild is required
        try:
            self._build_bm25(self._wiki_dir)
        except Exception:
            logger.debug("BM25 rebuild failed after update", exc_info=True)

    def search(self, query: str, top_k: int = 5) -> list[SearchResult]:
        """BM25 search; delegates to keyword fallback if bm25s unavailable."""
        if self._fallback is not None:
            return self._fallback.search(query, top_k)
        try:
            return self._search_bm25(query, top_k)
        except Exception:
            logger.debug("BM25 search error; rebuilding index", exc_info=True)
            self.build(self._wiki_dir)
            if self._fallback is not None:
                return self._fallback.search(query, top_k)
            try:
                return self._search_bm25(query, top_k)
            except Exception:
                logger.debug("BM25 search failed after rebuild", exc_info=True)
                return []

    # ------------------------------------------------------------------
    # Private helpers
    # ------------------------------------------------------------------

    def _build_bm25(self, wiki_dir: Path) -> None:
        import bm25s

        pages = _collect_pages(wiki_dir)
        self._pages = pages

        if not pages:
            logger.debug("BM25: wiki is empty — skipping index build")
            return

        corpus = [content for _, content in pages]
        tokenized = bm25s.tokenize(corpus)

        retriever = bm25s.BM25()
        retriever.index(tokenized)

        self._index_path.mkdir(parents=True, exist_ok=True)
        retriever.save(str(self._index_path))
        logger.info("BM25 index built: %d pages → %s", len(pages), self._index_path)

    def _search_bm25(self, query: str, top_k: int) -> list[SearchResult]:
        import bm25s

        if not self._index_path.exists():
            self._build_bm25(self._wiki_dir)

        retriever = bm25s.BM25.load(str(self._index_path), load_corpus=False)
        pages = self._pages or _collect_pages(self._wiki_dir)
        if not pages:
            return []

        k = min(top_k, len(pages))
        tokenized_query = bm25s.tokenize([query])
        results, scores = retriever.retrieve(tokenized_query, k=k)

        from .manager import parse_page

        output: list[SearchResult] = []
        for idx, score in zip(results[0], scores[0], strict=True):
            page_path, content = pages[int(idx)]
            fm, body = parse_page(content)
            output.append(
                SearchResult(
                    page_path=str(page_path.relative_to(self._wiki_dir)),
                    score=float(score),
                    excerpt=_excerpt(body),
                    frontmatter=fm,
                )
            )
        return output
build(wiki_dir)

Build BM25 index from all wiki pages, persist to disk.

Source code in src/dv_agentic/wiki/search.py 
175
176
177
178
179
180
181
182
183
184
185
def build(self, wiki_dir: Path) -> None:
    """Build BM25 index from all wiki pages, persist to disk."""
    try:
        self._build_bm25(wiki_dir)
    except ImportError:
        logger.info(
            "bm25s not installed — wiki search uses keyword fallback. "
            "Install with: pip install 'bm25s[core]'"
        )
        self._fallback = KeywordSearchIndex(wiki_dir)
        self._fallback.build(wiki_dir)
search(query, top_k=5)

BM25 search; delegates to keyword fallback if bm25s unavailable.

Source code in src/dv_agentic/wiki/search.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def search(self, query: str, top_k: int = 5) -> list[SearchResult]:
    """BM25 search; delegates to keyword fallback if bm25s unavailable."""
    if self._fallback is not None:
        return self._fallback.search(query, top_k)
    try:
        return self._search_bm25(query, top_k)
    except Exception:
        logger.debug("BM25 search error; rebuilding index", exc_info=True)
        self.build(self._wiki_dir)
        if self._fallback is not None:
            return self._fallback.search(query, top_k)
        try:
            return self._search_bm25(query, top_k)
        except Exception:
            logger.debug("BM25 search failed after rebuild", exc_info=True)
            return []
update(page_path, content)

Rebuild the full index after a page is written.

Source code in src/dv_agentic/wiki/search.py 
187
188
189
190
191
192
193
194
195
196
def update(self, page_path: Path, content: str) -> None:
    """Rebuild the full index after a page is written."""
    if self._fallback is not None:
        self._fallback.update(page_path, content)
        return
    # bm25s v0.2: no incremental API — full rebuild is required
    try:
        self._build_bm25(self._wiki_dir)
    except Exception:
        logger.debug("BM25 rebuild failed after update", exc_info=True)
KeywordSearchIndex

Bases: WikiSearchIndex

Simple in-memory keyword search — counts query term occurrences.

Used as a fallback when bm25s is not installed, and as the default when search_backend: "none" is set in project.yaml.

Source code in src/dv_agentic/wiki/search.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
class KeywordSearchIndex(WikiSearchIndex):
    """Simple in-memory keyword search — counts query term occurrences.

    Used as a fallback when ``bm25s`` is not installed, and as the default
    when ``search_backend: "none"`` is set in ``project.yaml``.
    """

    def __init__(self, wiki_dir: Path) -> None:
        self._wiki_dir = wiki_dir
        self._pages: list[tuple[Path, str]] = []

    def build(self, wiki_dir: Path) -> None:
        self._pages = _collect_pages(wiki_dir)
        logger.debug("Keyword index built: %d pages", len(self._pages))

    def update(self, page_path: Path, content: str) -> None:
        for i, (path, _) in enumerate(self._pages):
            if path == page_path:
                self._pages[i] = (page_path, content)
                return
        self._pages.append((page_path, content))

    def search(self, query: str, top_k: int = 5) -> list[SearchResult]:
        if not self._pages:
            self._pages = _collect_pages(self._wiki_dir)

        from .manager import parse_page

        query_lower = query.lower()
        words = [w for w in re.split(r"\W+", query_lower) if w]

        scored: list[tuple[float, Path, str]] = []
        for page_path, content in self._pages:
            cl = content.lower()
            score = float(sum(cl.count(w) for w in words))
            if score > 0:
                scored.append((score, page_path, content))

        scored.sort(key=lambda x: x[0], reverse=True)

        results: list[SearchResult] = []
        for score, page_path, content in scored[:top_k]:
            fm, body = parse_page(content)
            results.append(
                SearchResult(
                    page_path=str(page_path.relative_to(self._wiki_dir)),
                    score=score,
                    excerpt=_excerpt(body),
                    frontmatter=fm,
                )
            )
        return results
SearchResult dataclass

Single result returned by :meth:WikiSearchIndex.search.

Attributes:

Name Type Description
page_path str

Path relative to wiki_dir.
score float

BM25 or keyword relevance score (higher = more relevant).
excerpt str

First 200 characters of the page body.
frontmatter dict[str, Any]

Parsed YAML frontmatter dict (may be empty).
Source code in src/dv_agentic/wiki/search.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@dataclass
class SearchResult:
    """Single result returned by :meth:`WikiSearchIndex.search`.

    Attributes:
        page_path: Path relative to *wiki_dir*.
        score: BM25 or keyword relevance score (higher = more relevant).
        excerpt: First 200 characters of the page body.
        frontmatter: Parsed YAML frontmatter dict (may be empty).
    """

    page_path: str
    score: float
    excerpt: str
    frontmatter: dict[str, Any]
WikiSearchIndex

Bases: ABC

Abstract search index interface.

Source code in src/dv_agentic/wiki/search.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class WikiSearchIndex(abc.ABC):
    """Abstract search index interface."""

    @abc.abstractmethod
    def build(self, wiki_dir: Path) -> None:
        """Scan *wiki_dir* and build a full index from scratch."""

    @abc.abstractmethod
    def update(self, page_path: Path, content: str) -> None:
        """Incrementally update the index entry for *page_path*."""

    @abc.abstractmethod
    def search(self, query: str, top_k: int = 5) -> list[SearchResult]:
        """Return up to *top_k* pages most relevant to *query*."""

    @classmethod
    def create(cls, backend: str, wiki_dir: Path) -> WikiSearchIndex:
        """Factory: instantiate the appropriate backend.

        Args:
            backend: ``"bm25"`` (recommended) or ``"none"`` (keyword only).
            wiki_dir: Wiki root directory.

        Returns:
            Concrete :class:`WikiSearchIndex` instance.

        Raises:
            ValueError: If *backend* is not recognised.
        """
        if backend == "bm25":
            return BM25SearchIndex(wiki_dir)
        if backend == "none":
            return KeywordSearchIndex(wiki_dir)
        raise ValueError(f"Unknown search backend: {backend!r}. Supported values: 'bm25', 'none'.")
build(wiki_dir) abstractmethod

Scan wiki_dir and build a full index from scratch.

Source code in src/dv_agentic/wiki/search.py 
81
82
83
@abc.abstractmethod
def build(self, wiki_dir: Path) -> None:
    """Scan *wiki_dir* and build a full index from scratch."""
create(backend, wiki_dir) classmethod

Factory: instantiate the appropriate backend.

Parameters:

Name Type Description Default
backend str

"bm25" (recommended) or "none" (keyword only).

 required
wiki_dir Path

Wiki root directory.

 required

Returns:

Name Type Description
Concrete WikiSearchIndex

class:WikiSearchIndex instance.

Raises:

Type Description
ValueError

If backend is not recognised.
Source code in src/dv_agentic/wiki/search.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@classmethod
def create(cls, backend: str, wiki_dir: Path) -> WikiSearchIndex:
    """Factory: instantiate the appropriate backend.

    Args:
        backend: ``"bm25"`` (recommended) or ``"none"`` (keyword only).
        wiki_dir: Wiki root directory.

    Returns:
        Concrete :class:`WikiSearchIndex` instance.

    Raises:
        ValueError: If *backend* is not recognised.
    """
    if backend == "bm25":
        return BM25SearchIndex(wiki_dir)
    if backend == "none":
        return KeywordSearchIndex(wiki_dir)
    raise ValueError(f"Unknown search backend: {backend!r}. Supported values: 'bm25', 'none'.")
search(query, top_k=5) abstractmethod

Return up to top_k pages most relevant to query.

Source code in src/dv_agentic/wiki/search.py 
89
90
91
@abc.abstractmethod
def search(self, query: str, top_k: int = 5) -> list[SearchResult]:
    """Return up to *top_k* pages most relevant to *query*."""
update(page_path, content) abstractmethod

Incrementally update the index entry for page_path.

Source code in src/dv_agentic/wiki/search.py 
85
86
87
@abc.abstractmethod
def update(self, page_path: Path, content: str) -> None:
    """Incrementally update the index entry for *page_path*."""

Wiki Build Command

wiki_build

Wiki build CLI.

Rebuilds index.md and the BM25 search index.

Usage

python -m dv_agentic.cli.wiki_build