Record the input side of a triple. Returns a triple_id.
(
self,
subsystem: str,
stage: str,
context: str,
*,
session_id: str = "",
workspace_id: str = "",
)
| 70 | return f"ram-{h}" |
| 71 | |
| 72 | def record_input( |
| 73 | self, |
| 74 | subsystem: str, |
| 75 | stage: str, |
| 76 | context: str, |
| 77 | *, |
| 78 | session_id: str = "", |
| 79 | workspace_id: str = "", |
| 80 | ) -> str: |
| 81 | """Record the input side of a triple. Returns a triple_id.""" |
| 82 | if not self.enabled: |
| 83 | return "" |
| 84 | triple_id = self._make_id(subsystem, stage, context) |
| 85 | triple = RAMTriple( |
| 86 | triple_id=triple_id, |
| 87 | subsystem=subsystem, |
| 88 | stage=stage, |
| 89 | x_context=context, |
| 90 | session_id=session_id, |
| 91 | workspace_id=workspace_id, |
| 92 | ) |
| 93 | self._pending[triple_id] = triple |
| 94 | return triple_id |
| 95 | |
| 96 | def record_output(self, triple_id: str, output: str) -> None: |
| 97 | """Attach RAM's generated output to a pending triple.""" |
no test coverage detected