Store an operation with full context
(self, op_type: str, parameters: Dict, context: Dict)
| 100 | return session_id |
| 101 | |
| 102 | def store_operation(self, op_type: str, parameters: Dict, context: Dict): |
| 103 | """Store an operation with full context""" |
| 104 | if not hasattr(self, 'current_session'): |
| 105 | self.start_session("Auto-generated session") |
| 106 | |
| 107 | cursor = self.conn.cursor() |
| 108 | cursor.execute(""" |
| 109 | INSERT INTO operations |
| 110 | (session_id, operation_type, parameters, context, timestamp, was_successful, time_taken) |
| 111 | VALUES (?, ?, ?, ?, ?, ?, ?) |
| 112 | """, ( |
| 113 | self.current_session, |
| 114 | op_type, |
| 115 | json.dumps(parameters), |
| 116 | json.dumps(context), |
| 117 | datetime.now(), |
| 118 | True, # Assume successful unless marked otherwise |
| 119 | context.get("duration", 0) |
| 120 | )) |
| 121 | |
| 122 | self.conn.commit() |
| 123 | |
| 124 | # Check for patterns |
| 125 | self._detect_and_store_patterns() |
| 126 | |
| 127 | def mark_operation_failed(self, operation_id: int): |
| 128 | """Mark an operation as failed/undone""" |
nothing calls this directly
no test coverage detected