Chronica

Instructions

Implementation Reference

• src/chronica/tools.py:399-417 (handler)

  Tool handler logic for "chronica_timeline" which calls the store's timeline method.

  elif name == "chronica_timeline":
      thread_type = arguments.get("thread_type")
      if thread_type and thread_type not in ["normal", "project"]:
          return [types.TextContent(
              type="text",
              text=json.dumps({"error": "invalid_thread", "message": f"thread_type must be 'normal' or 'project', got: {thread_type}"}, ensure_ascii=False)
          )]
      
      entries = store.timeline(
          start_time=arguments.get("start_time"),
          end_time=arguments.get("end_time"),
          thread_type=thread_type,
          kind=arguments.get("kind"),
          limit=arguments.get("limit", 100)
      )
      return [types.TextContent(
          type="text",
          text=json.dumps({"entries": entries}, ensure_ascii=False, indent=2)
      )]

• src/chronica/store.py:235-276 (handler)

  The actual implementation in the data store which queries the sqlite database for timeline entries.

  def timeline(
      self,
      start_time: Optional[str] = None,
      end_time: Optional[str] = None,
      thread_id: Optional[str] = None,
      thread_type: Optional[str] = None,
      kind: Optional[str] = None,
      limit: int = 100
  ) -> List[Dict[str, Any]]:
      """タイムラインを取得"""
      conn = sqlite3.connect(str(self.db_path))
      conn.row_factory = sqlite3.Row
      cursor = conn.cursor()
      
      conditions = []
      params = []
      
      if start_time:
          conditions.append("saved_time >= ?")
          params.append(start_time)
      if end_time:
          conditions.append("saved_time <= ?")
          params.append(end_time)
      if thread_id:
          conditions.append("thread_id = ?")
          params.append(thread_id)
      if thread_type:
          conditions.append("thread_type = ?")
          params.append(thread_type)
      if kind:
          conditions.append("kind = ?")
          params.append(kind)
      
      where_clause = " AND ".join(conditions) if conditions else "1=1"
      query = f"SELECT * FROM entries WHERE {where_clause} ORDER BY saved_time ASC LIMIT ?"
      params.append(limit)
      
      cursor.execute(query, params)
      rows = cursor.fetchall()
      conn.close()
      
      return [self._row_to_entry(row) for row in rows]