query_trace_file

Instructions

Implementation Reference

• src/camoufox_reverse_mcp/tools/trace.py:164-201 (handler)

  Main handler function for the 'query_trace_file' tool. Loads a .jsonl trace file, filters events by object/search, and returns results in summary/timeline/sequence mode.

  @mcp.tool()
  async def query_trace_file(
      file_path: str,
      mode: str = "summary",
      filter_object: Optional[str] = None,
      search_query: Optional[str] = None,
      limit: int = 1000,
      bucket_ms: int = 500,
  ) -> dict:
      """Query a specific historical trace file (post-hoc analysis).
  
      Args:
          file_path: Path to the .jsonl trace file.
          mode: Same as trace_property_access (summary/timeline/sequence/search).
          filter_object: Filter by object name.
          search_query: Filter by search string.
          limit: Max events for sequence mode.
          bucket_ms: Bucket size for timeline mode.
      """
      path = Path(file_path)
      if not path.exists():
          return {"mode": "error", "reason": f"File not found: {file_path}"}
  
      events = load_events(path)
      events = filter_events(events, filter_object, search_query)
  
      duration_s = 0
      if events:
          duration_s = (events[-1].get("t", 0) // 1000) + 1
  
      if mode == "summary":
          return build_summary(events, duration_s)
      elif mode == "timeline":
          return build_timeline(events, duration_s, bucket_ms)
      elif mode in ("sequence", "search"):
          return build_sequence(events, limit)
      else:
          return {"mode": "error", "reason": f"Unknown mode: {mode}"}

• src/camoufox_reverse_mcp/tools/trace.py:164-165 (registration)

  The tool is registered via @mcp.tool() decorator on the query_trace_file function.

  @mcp.tool()
  async def query_trace_file(

• src/camoufox_reverse_mcp/property_trace.py:79-92 (helper)

  load_events helper: reads and parses JSONL trace events from a file.

  def load_events(jsonl_path: Path) -> list[dict]:
      events = []
      if not jsonl_path.exists():
          return events
      with open(jsonl_path, "r", encoding="utf-8") as f:
          for line in f:
              line = line.strip()
              if not line:
                  continue
              try:
                  events.append(json.loads(line))
              except json.JSONDecodeError:
                  continue
      return events

• src/camoufox_reverse_mcp/property_trace.py:201-216 (helper)

  filter_events helper: filters events by object name and search query.

  def filter_events(
      events: list[dict],
      filter_object: Optional[str] = None,
      search_query: Optional[str] = None,
  ) -> list[dict]:
      if filter_object:
          events = [e for e in events if e.get("o") == filter_object]
      if search_query:
          q = search_query.lower()
          events = [
              e for e in events
              if q in str(e.get("p", "")).lower()
              or q in str(e.get("v", "")).lower()
              or q in str(e.get("o", "")).lower()
          ]
      return events

• src/camoufox_reverse_mcp/property_trace.py:119-198 (helper)

  build_summary, build_timeline, build_sequence helpers: aggregate trace events into different view modes.

  def build_summary(events: list[dict], duration_s: int) -> dict:
      by_path: dict[str, dict] = defaultdict(lambda: {
          "count": 0, "first_ms": None, "last_ms": None,
      })
      by_object: dict[str, int] = defaultdict(int)
  
      for e in events:
          obj = e.get("o", "")
          prop = e.get("p", "")
          path = f"{obj}.{prop}"
          ts = e.get("t", 0)
  
          entry = by_path[path]
          entry["count"] += 1
          if entry["first_ms"] is None or ts < entry["first_ms"]:
              entry["first_ms"] = ts
          if entry["last_ms"] is None or ts > entry["last_ms"]:
              entry["last_ms"] = ts
          by_object[obj] += 1
  
      by_property_list = [
          {"path": path, **stats}
          for path, stats in sorted(by_path.items(), key=lambda x: -x[1]["count"])
      ]
  
      return {
          "mode": "summary",
          "duration_s": duration_s,
          "total_events": len(events),
          "unique_properties": len(by_path),
          "by_property": by_property_list,
          "by_object": dict(sorted(by_object.items(), key=lambda x: -x[1])),
      }
  
  
  def build_timeline(events: list[dict], duration_s: int, bucket_ms: int) -> dict:
      if not events:
          return {"mode": "timeline", "duration_s": duration_s,
                  "bucket_ms": bucket_ms, "buckets": []}
  
      max_ms = max(e.get("t", 0) for e in events)
      n_buckets = (max_ms // bucket_ms) + 1
      buckets = [
          {"from_ms": i * bucket_ms, "to_ms": (i + 1) * bucket_ms,
           "events": 0, "new_properties": []}
          for i in range(n_buckets)
      ]
  
      seen: set[str] = set()
      for e in events:
          ts = e.get("t", 0)
          idx = ts // bucket_ms
          if idx >= n_buckets:
              continue
          path = f"{e.get('o', '')}.{e.get('p', '')}"
          buckets[idx]["events"] += 1
          if path not in seen:
              seen.add(path)
              buckets[idx]["new_properties"].append(path)
  
      return {"mode": "timeline", "duration_s": duration_s,
              "bucket_ms": bucket_ms, "buckets": buckets}
  
  
  def build_sequence(events: list[dict], limit: int) -> dict:
      truncated = len(events) > limit
      shown = events[:limit]
      return {
          "mode": "sequence",
          "total_events": len(events),
          "returned": len(shown),
          "truncated": truncated,
          "events": [
              {"idx": i, "ms": e.get("t", 0),
               "path": f"{e.get('o', '')}.{e.get('p', '')}",
               "kind": {0: "get", 1: "set", 2: "call"}.get(e.get("k", 0), "?"),
               "v": e.get("v", "")}
              for i, e in enumerate(shown)
          ],
      }